You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ec...@apache.org on 2012/04/06 18:35:31 UTC
svn commit: r1310457 [2/3] - in /accumulo/trunk:
core/src/main/java/org/apache/accumulo/core/client/impl/
core/src/main/java/org/apache/accumulo/core/conf/
core/src/main/java/org/apache/accumulo/core/data/
server/src/main/java/org/apache/accumulo/serve...
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tserverOps/StopLogger.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tserverOps/StopLogger.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tserverOps/StopLogger.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tserverOps/StopLogger.java Fri Apr 6 16:35:29 2012
@@ -21,7 +21,6 @@ import java.net.InetSocketAddress;
import org.apache.accumulo.core.tabletserver.thrift.MutationLogger;
import org.apache.accumulo.core.util.AddressUtil;
import org.apache.accumulo.core.util.ThriftUtil;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.server.fate.Repo;
import org.apache.accumulo.server.master.Master;
import org.apache.accumulo.server.master.tableOps.MasterRepo;
@@ -46,7 +45,7 @@ public class StopLogger extends MasterRe
@Override
public Repo<Master> call(long tid, Master m) throws Exception {
InetSocketAddress addr = AddressUtil.parseAddress(logger, 0);
- MutationLogger.Iface client = ThriftUtil.getClient(new MutationLogger.Client.Factory(), addr, ServerConfiguration.getSystemConfiguration());
+ MutationLogger.Iface client = ThriftUtil.getClient(new MutationLogger.Client.Factory(), addr, m.getSystemConfiguration());
try {
client.halt(null, SecurityConstants.getSystemCredentials());
log.info("logger asked to halt " + logger);
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/LogService.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/LogService.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/LogService.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/LogService.java Fri Apr 6 16:35:29 2012
@@ -24,9 +24,9 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.util.Daemon;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apache.log4j.net.SocketNode;
@@ -70,9 +70,9 @@ public class LogService extends org.apac
}
}
- static void startLogListener() {
+ static void startLogListener(AccumuloConfiguration conf) {
try {
- new Daemon(new SocketServer(ServerConfiguration.getSystemConfiguration().getPort(Property.MONITOR_LOG4J_PORT))).start();
+ new Daemon(new SocketServer(conf.getPort(Property.MONITOR_LOG4J_PORT))).start();
} catch (Throwable t) {
log.info("Unable to listen to cluster-wide ports", t);
}
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/Monitor.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/Monitor.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/Monitor.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/Monitor.java Fri Apr 6 16:35:29 2012
@@ -30,6 +30,7 @@ import java.util.Set;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.MasterClient;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.file.FileUtil;
import org.apache.accumulo.core.gc.thrift.GCMonitorService;
@@ -141,6 +142,8 @@ public class Monitor {
private static Instance instance;
+ private static ServerConfiguration config;
+
public static Map<String,Double> summarizeTableStats(MasterMonitorInfo mmi) {
Map<String,Double> compactingByTable = new HashMap<String,Double>();
if (mmi != null && mmi.tServerInfo != null) {
@@ -399,7 +402,7 @@ public class Monitor {
Instance instance = HdfsZooInstance.getInstance();
String zooKeepers = instance.getZooKeepers();
log.debug("connecting to zookeepers " + zooKeepers);
- ZooKeeper zk = new ZooKeeper(zooKeepers, (int) ServerConfiguration.getSystemConfiguration().getTimeInMillis(Property.INSTANCE_ZK_TIMEOUT), new Watcher() {
+ ZooKeeper zk = new ZooKeeper(zooKeepers, (int) config.getConfiguration().getTimeInMillis(Property.INSTANCE_ZK_TIMEOUT), new Watcher() {
@Override
public void process(WatchedEvent event) {}
});
@@ -409,7 +412,7 @@ public class Monitor {
if (locks != null && locks.size() > 0) {
Collections.sort(locks);
InetSocketAddress address = new ServerServices(new String(zk.getData(path + "/" + locks.get(0), null, null))).getAddress(Service.GC_CLIENT);
- GCMonitorService.Iface client = ThriftUtil.getClient(new GCMonitorService.Client.Factory(), address, ServerConfiguration.getSystemConfiguration());
+ GCMonitorService.Iface client = ThriftUtil.getClient(new GCMonitorService.Client.Factory(), address, config.getConfiguration());
try {
result = client.getStatus(null, SecurityConstants.getSystemCredentials());
} finally {
@@ -427,9 +430,10 @@ public class Monitor {
public static void main(String[] args) throws Exception {
FileSystem fs = FileUtil.getFileSystem(CachedConfiguration.getInstance(), ServerConfiguration.getSiteConfiguration());
- Accumulo.init(fs, "monitor");
String hostname = Accumulo.getLocalAddress(args);
instance = HdfsZooInstance.getInstance();
+ config = new ServerConfiguration(instance);
+ Accumulo.init(fs, config, "monitor");
Monitor monitor = new Monitor();
Accumulo.enableTracing(hostname, "monitor");
monitor.run(hostname);
@@ -439,7 +443,7 @@ public class Monitor {
public void run(String hostname) {
Monitor.START_TIME = System.currentTimeMillis();
- int port = ServerConfiguration.getSystemConfiguration().getPort(Property.MONITOR_PORT);
+ int port = config.getConfiguration().getPort(Property.MONITOR_PORT);
EmbeddedWebServer server;
try {
log.debug("Creating monitor on port " + port);
@@ -463,7 +467,7 @@ public class Monitor {
server.addServlet(Summary.class, "/trace/summary");
server.addServlet(ListType.class, "/trace/listType");
server.addServlet(ShowTrace.class, "/trace/show");
- LogService.startLogListener();
+ LogService.startLogListener(Monitor.getSystemConfiguration());
server.start();
new Daemon(new LoggingRunnable(log, new ZooKeeperStatus()), "ZooKeeperStatus").start();
@@ -616,4 +620,11 @@ public class Monitor {
return new ArrayList<Pair<Long,Double>>(dataCacheHitRateOverTime);
}
}
+
+ /**
+ * @return
+ */
+ public static AccumuloConfiguration getSystemConfiguration() {
+ return config.getConfiguration();
+ }
}
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/ZooKeeperStatus.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/ZooKeeperStatus.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/ZooKeeperStatus.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/ZooKeeperStatus.java Fri Apr 6 16:35:29 2012
@@ -65,7 +65,7 @@ public class ZooKeeperStatus implements
List<ZooKeeperState> update = new ArrayList<ZooKeeperState>();
- String zookeepers[] = ServerConfiguration.getSystemConfiguration().get(Property.INSTANCE_ZK_HOST).split(",");
+ String zookeepers[] = ServerConfiguration.getSiteConfiguration().get(Property.INSTANCE_ZK_HOST).split(",");
for (String keeper : zookeepers) {
int clients = 0;
String mode = "unknown";
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/BasicServlet.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/BasicServlet.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/BasicServlet.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/BasicServlet.java Fri Apr 6 16:35:29 2012
@@ -34,7 +34,6 @@ import javax.servlet.http.HttpServletRes
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.server.client.HdfsZooInstance;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.server.monitor.LogService;
import org.apache.accumulo.server.monitor.Monitor;
import org.apache.accumulo.server.util.time.SimpleTimer;
@@ -55,9 +54,9 @@ abstract public class BasicServlet exten
StringBuilder sb = new StringBuilder();
try {
Monitor.fetchData();
- bannerText = sanitize(ServerConfiguration.getSystemConfiguration().get(Property.MONITOR_BANNER_TEXT));
- bannerColor = ServerConfiguration.getSystemConfiguration().get(Property.MONITOR_BANNER_COLOR).replace("'", "'");
- bannerBackground = ServerConfiguration.getSystemConfiguration().get(Property.MONITOR_BANNER_BACKGROUND).replace("'", "'");
+ bannerText = sanitize(Monitor.getSystemConfiguration().get(Property.MONITOR_BANNER_TEXT));
+ bannerColor = Monitor.getSystemConfiguration().get(Property.MONITOR_BANNER_COLOR).replace("'", "'");
+ bannerBackground = Monitor.getSystemConfiguration().get(Property.MONITOR_BANNER_BACKGROUND).replace("'", "'");
pageStart(req, resp, sb);
pageBody(req, resp, sb);
pageEnd(req, resp, sb);
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/DefaultServlet.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/DefaultServlet.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/DefaultServlet.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/DefaultServlet.java Fri Apr 6 16:35:29 2012
@@ -266,7 +266,7 @@ public class DefaultServlet extends Basi
String consumed = "Unknown";
String diskUsed = "Unknown";
try {
- Path path = new Path(ServerConfiguration.getSystemConfiguration().get(Property.INSTANCE_DFS_DIR));
+ Path path = new Path(Monitor.getSystemConfiguration().get(Property.INSTANCE_DFS_DIR));
log.debug("Reading the content summary for " + path);
ContentSummary acu = fs.getContentSummary(path);
consumed = String.format("%.2f%%", acu.getSpaceConsumed() * 100. / fs.getUsed());
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/MasterServlet.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/MasterServlet.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/MasterServlet.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/MasterServlet.java Fri Apr 6 16:35:29 2012
@@ -36,7 +36,6 @@ import org.apache.accumulo.core.master.t
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
import org.apache.accumulo.core.util.StringUtil;
import org.apache.accumulo.server.client.HdfsZooInstance;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.server.monitor.DedupedLogEvent;
import org.apache.accumulo.server.monitor.LogService;
import org.apache.accumulo.server.monitor.Monitor;
@@ -107,7 +106,7 @@ public class MasterServlet extends Basic
long diff = System.currentTimeMillis() - start;
gcStatus = label + " " + DateFormat.getInstance().format(new Date(start));
gcStatus = gcStatus.replace(" ", " ");
- long normalDelay = ServerConfiguration.getSystemConfiguration().getTimeInMillis(Property.GC_CYCLE_DELAY);
+ long normalDelay = Monitor.getSystemConfiguration().getTimeInMillis(Property.GC_CYCLE_DELAY);
if (diff > normalDelay * 2)
gcStatus = "<span class='warning'>" + gcStatus + "</span>";
}
@@ -168,7 +167,6 @@ public class MasterServlet extends Basic
private void doRecoveryList(HttpServletRequest req, StringBuilder sb) {
MasterMonitorInfo mmi = Monitor.getMmi();
- boolean sortingUsesMapReduce = ServerConfiguration.getSystemConfiguration().getBoolean(Property.MASTER_RECOVERY_SORT_MAPREDUCE);
if (mmi != null) {
List<RecoveryStatus> jobs = mmi.recovery;
if (jobs != null && jobs.size() > 0) {
@@ -177,13 +175,7 @@ public class MasterServlet extends Basic
recoveryTable.addSortableColumn("Server", new LoggerLinkType(), null);
recoveryTable.addSortableColumn("Log");
recoveryTable.addSortableColumn("Time", new DurationType(), null);
- if (sortingUsesMapReduce) {
- recoveryTable.addSortableColumn("Copy", new ProgressChartType(), null);
- recoveryTable.addSortableColumn("Map", new ProgressChartType(), null);
- recoveryTable.addSortableColumn("Reduce", new ProgressChartType(), null);
- } else {
- recoveryTable.addSortableColumn("Copy/Sort", new ProgressChartType(), null);
- }
+ recoveryTable.addSortableColumn("Copy/Sort", new ProgressChartType(), null);
for (RecoveryStatus recovery : jobs) {
TableRow row = recoveryTable.prepareRow();
@@ -191,10 +183,6 @@ public class MasterServlet extends Basic
row.add(recovery.name);
row.add((long) recovery.runtime);
row.add(recovery.copyProgress);
- if (sortingUsesMapReduce) {
- row.add(recovery.mapProgress);
- row.add(recovery.reduceProgress);
- }
recoveryTable.addRow(row);
}
recoveryTable.generate(req, sb);
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/TServersServlet.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/TServersServlet.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/TServersServlet.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/TServersServlet.java Fri Apr 6 16:35:29 2012
@@ -38,7 +38,6 @@ import org.apache.accumulo.core.tabletse
import org.apache.accumulo.core.util.AddressUtil;
import org.apache.accumulo.core.util.Duration;
import org.apache.accumulo.core.util.ThriftUtil;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.server.master.state.TabletServerState;
import org.apache.accumulo.server.monitor.Monitor;
import org.apache.accumulo.server.monitor.util.Table;
@@ -123,7 +122,7 @@ public class TServersServlet extends Bas
TabletStats historical = new TabletStats(null, new ActionStats(), new ActionStats(), new ActionStats(), 0, 0, 0, 0);
List<TabletStats> tsStats = new ArrayList<TabletStats>();
try {
- TabletClientService.Iface client = ThriftUtil.getClient(new TabletClientService.Client.Factory(), address, ServerConfiguration.getSystemConfiguration());
+ TabletClientService.Iface client = ThriftUtil.getClient(new TabletClientService.Client.Factory(), address, Monitor.getSystemConfiguration());
try {
for (String tableId : Monitor.getMmi().tableMap.keySet()) {
tsStats.addAll(client.getTabletStats(null, SecurityConstants.getSystemCredentials(), tableId));
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/VisServlet.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/VisServlet.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/VisServlet.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/VisServlet.java Fri Apr 6 16:35:29 2012
@@ -26,11 +26,10 @@ import javax.servlet.http.HttpServletRes
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.server.monitor.Monitor;
public class VisServlet extends BasicServlet {
- private static final int concurrentScans = ServerConfiguration.getSystemConfiguration().getCount(Property.TSERV_READ_AHEAD_MAXCONCURRENT);
+ private static final int concurrentScans = Monitor.getSystemConfiguration().getCount(Property.TSERV_READ_AHEAD_MAXCONCURRENT);
private static final long serialVersionUID = 1L;
boolean useCircles;
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/trace/Basic.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/trace/Basic.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/trace/Basic.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/monitor/servlets/trace/Basic.java Fri Apr 6 16:35:29 2012
@@ -30,7 +30,7 @@ import org.apache.accumulo.core.conf.Pro
import org.apache.accumulo.core.trace.TraceDump;
import org.apache.accumulo.core.trace.TraceFormatter;
import org.apache.accumulo.server.client.HdfsZooInstance;
-import org.apache.accumulo.server.conf.ServerConfiguration;
+import org.apache.accumulo.server.monitor.Monitor;
import org.apache.accumulo.server.monitor.servlets.BasicServlet;
abstract class Basic extends BasicServlet {
@@ -63,7 +63,7 @@ abstract class Basic extends BasicServle
}
protected Scanner getScanner(StringBuilder sb) throws AccumuloException {
- AccumuloConfiguration conf = ServerConfiguration.getSystemConfiguration();
+ AccumuloConfiguration conf = Monitor.getSystemConfiguration();
String user = conf.get(Property.TRACE_USER);
byte[] passwd = conf.get(Property.TRACE_PASSWORD).getBytes();
String table = conf.get(Property.TRACE_TABLE);
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/security/SecurityConstants.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/security/SecurityConstants.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/security/SecurityConstants.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/security/SecurityConstants.java Fri Apr 6 16:35:29 2012
@@ -134,7 +134,7 @@ public class SecurityConstants {
md.update(Constants.VERSION.getBytes());
md.update(HdfsZooInstance.getInstance().getInstanceID().getBytes());
- for (Entry<String,String> entry : ServerConfiguration.getSystemConfiguration()) {
+ for (Entry<String,String> entry : ServerConfiguration.getSiteConfiguration()) {
// only include instance properties
if (entry.getKey().startsWith(Property.INSTANCE_PREFIX.toString())) {
md.update(entry.getKey().getBytes());
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/Compactor.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/Compactor.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/Compactor.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/Compactor.java Fri Apr 6 16:35:29 2012
@@ -27,7 +27,6 @@ import java.util.concurrent.Callable;
import org.apache.accumulo.cloudtrace.instrument.Span;
import org.apache.accumulo.cloudtrace.instrument.Trace;
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.KeyExtent;
@@ -36,8 +35,8 @@ import org.apache.accumulo.core.file.Fil
import org.apache.accumulo.core.file.FileSKVIterator;
import org.apache.accumulo.core.file.FileSKVWriter;
import org.apache.accumulo.core.iterators.IteratorUtil;
-import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iterators.system.CountingIterator;
import org.apache.accumulo.core.iterators.system.DeletingIterator;
import org.apache.accumulo.core.iterators.system.MultiIterator;
@@ -45,7 +44,6 @@ import org.apache.accumulo.core.iterator
import org.apache.accumulo.core.util.LocalityGroupUtil;
import org.apache.accumulo.core.util.LocalityGroupUtil.LocalityGroupConfigurationError;
import org.apache.accumulo.core.util.MetadataTable.DataFileValue;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.server.conf.TableConfiguration;
import org.apache.accumulo.server.problems.ProblemReport;
import org.apache.accumulo.server.problems.ProblemReportingIterator;
@@ -115,12 +113,11 @@ public class Compactor implements Callab
try {
FileOperations fileFactory = FileOperations.getInstance();
- AccumuloConfiguration tableConf = ServerConfiguration.getTableConfiguration(extent.getTableId().toString());
- mfw = fileFactory.openWriter(outputFile, fs, conf, tableConf);
+ mfw = fileFactory.openWriter(outputFile, fs, conf, acuTableConf);
Map<String,Set<ByteSequence>> lGroups;
try {
- lGroups = LocalityGroupUtil.getLocalityGroups(tableConf);
+ lGroups = LocalityGroupUtil.getLocalityGroups(acuTableConf);
} catch (LocalityGroupConfigurationError e) {
throw new IOException(e);
}
@@ -146,7 +143,7 @@ public class Compactor implements Callab
// Verify the file, since hadoop 0.20.2 sometimes lies about the success of close()
try {
- FileSKVIterator openReader = fileFactory.openReader(outputFile, false, fs, conf, tableConf);
+ FileSKVIterator openReader = fileFactory.openReader(outputFile, false, fs, conf, acuTableConf);
openReader.close();
} catch (IOException ex) {
log.error("Verification of successful compaction fails!!! " + extent + " " + outputFile, ex);
@@ -156,7 +153,7 @@ public class Compactor implements Callab
log.debug(String.format("Compaction %s %,d read | %,d written | %,6d entries/sec | %6.3f secs", extent, majCStats.getEntriesRead(),
majCStats.getEntriesWritten(), (int) (majCStats.getEntriesRead() / ((t2 - t1) / 1000.0)), (t2 - t1) / 1000.0));
- majCStats.setFileSize(fileFactory.getFileSize(outputFile, fs, conf, ServerConfiguration.getTableConfiguration(extent.getTableId().toString())));
+ majCStats.setFileSize(fileFactory.getFileSize(outputFile, fs, conf, acuTableConf));
return majCStats;
} catch (IOException e) {
log.error(e, e);
@@ -194,9 +191,7 @@ public class Compactor implements Callab
FileSKVIterator reader;
- AccumuloConfiguration tableConf = ServerConfiguration.getTableConfiguration(extent.getTableId().toString());
-
- reader = fileFactory.openReader(mapFile, false, fs, conf, tableConf);
+ reader = fileFactory.openReader(mapFile, false, fs, conf, acuTableConf);
readers.add(reader);
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/FileManager.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/FileManager.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/FileManager.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/FileManager.java Fri Apr 6 16:35:29 2012
@@ -29,7 +29,6 @@ import java.util.Map.Entry;
import java.util.TimerTask;
import java.util.concurrent.Semaphore;
-import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.KeyExtent;
@@ -112,6 +111,8 @@ public class FileManager {
private long maxIdleTime;
+ private final ServerConfiguration conf;
+
private class IdleFileCloser extends TimerTask {
@Override
@@ -153,6 +154,7 @@ public class FileManager {
/**
*
+ * @param instance
* @param conf
* @param fs
* @param maxOpen
@@ -161,11 +163,11 @@ public class FileManager {
* @param indexCache
* : underlying file can and should be able to handle a null cache
*/
- FileManager(FileSystem fs, int maxOpen, BlockCache dataCache, BlockCache indexCache) {
+ FileManager(ServerConfiguration conf, FileSystem fs, int maxOpen, BlockCache dataCache, BlockCache indexCache) {
if (maxOpen <= 0)
throw new IllegalArgumentException("maxOpen <= 0");
-
+ this.conf = conf;
this.dataCache = dataCache;
this.indexCache = indexCache;
@@ -176,7 +178,7 @@ public class FileManager {
this.openFiles = new HashMap<String,List<OpenReader>>();
this.reservedReaders = new HashMap<FileSKVIterator,String>();
- this.maxIdleTime = ServerConfiguration.getSystemConfiguration().getTimeInMillis(Property.TSERV_MAX_IDLE);
+ this.maxIdleTime = conf.getConfiguration().getTimeInMillis(Property.TSERV_MAX_IDLE);
SimpleTimer.getInstance().schedule(new IdleFileCloser(), maxIdleTime, maxIdleTime / 2);
}
@@ -308,8 +310,7 @@ public class FileManager {
for (String file : filesToOpen) {
try {
// log.debug("Opening "+file);
- FileSKVIterator reader = FileOperations.getInstance().openReader(file, false, fs, fs.getConf(),
- ServerConfiguration.getTableConfiguration(table.toString()),
+ FileSKVIterator reader = FileOperations.getInstance().openReader(file, false, fs, fs.getConf(), conf.getTableConfiguration(table.toString()),
dataCache, indexCache);
reservedFiles.add(reader);
readersReserved.put(reader, file);
@@ -450,9 +451,9 @@ public class FileManager {
dataSources = new ArrayList<FileDataSource>();
this.tablet = tablet;
- continueOnFailure = ServerConfiguration.getTableConfiguration(tablet.getTableId().toString()).getBoolean(Property.TABLE_FAILURES_IGNORE);
+ continueOnFailure = conf.getTableConfiguration(tablet).getBoolean(Property.TABLE_FAILURES_IGNORE);
- if (tablet.getTableId().toString().equals(Constants.METADATA_TABLE_ID)) {
+ if (tablet.isMeta()) {
continueOnFailure = false;
}
}
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/InMemoryMap.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/InMemoryMap.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/InMemoryMap.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/InMemoryMap.java Fri Apr 6 16:35:29 2012
@@ -34,6 +34,7 @@ import java.util.concurrent.atomic.Atomi
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.ColumnUpdate;
@@ -209,9 +210,8 @@ public class InMemoryMap {
}
}
- public InMemoryMap() {
- this(ServerConfiguration.getSystemConfiguration().getBoolean(Property.TSERV_NATIVEMAP_ENABLED), ServerConfiguration.getSystemConfiguration().get(
- Property.TSERV_MEMDUMP_DIR));
+ public InMemoryMap(AccumuloConfiguration config) {
+ this(config.getBoolean(Property.TSERV_NATIVEMAP_ENABLED), config.get(Property.TSERV_MEMDUMP_DIR));
}
private interface SimpleMap {
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/LargestFirstMemoryManager.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/LargestFirstMemoryManager.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/LargestFirstMemoryManager.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/LargestFirstMemoryManager.java Fri Apr 6 16:35:29 2012
@@ -31,32 +31,41 @@ public class LargestFirstMemoryManager i
private static final Logger log = Logger.getLogger(LargestFirstMemoryManager.class);
- private final long maxMemory;
- private final int maxConcurrentMincs;
- private final int numWaitingMultiplier;
+ private long maxMemory = -1;
+ private int maxConcurrentMincs;
+ private int numWaitingMultiplier;
private long prevIngestMemory;
private double compactionThreshold;
private long maxObserved;
private HashMap<Text,Long> mincIdleThresholds;
private static final long zerotime = System.currentTimeMillis();
+ private ServerConfiguration config = null;
LargestFirstMemoryManager(long maxMemory, int maxConcurrentMincs, int numWaitingMultiplier) {
+ this();
this.maxMemory = maxMemory;
this.maxConcurrentMincs = maxConcurrentMincs;
this.numWaitingMultiplier = numWaitingMultiplier;
+ }
+
+ public void init(ServerConfiguration conf) {
+ this.config = conf;
+ maxMemory = conf.getConfiguration().getMemoryInBytes(Property.TSERV_MAXMEM);
+ maxConcurrentMincs = conf.getConfiguration().getCount(Property.TSERV_MINC_MAXCONCURRENT);
+ numWaitingMultiplier = Constants.TSERV_MINC_MAXCONCURRENT_NUMWAITING_MULTIPLIER;
+ }
+
+ LargestFirstMemoryManager() {
prevIngestMemory = 0;
compactionThreshold = 0.5;
maxObserved = 0;
mincIdleThresholds = new HashMap<Text,Long>();
}
- LargestFirstMemoryManager() {
- this(ServerConfiguration.getSystemConfiguration().getMemoryInBytes(Property.TSERV_MAXMEM), ServerConfiguration.getSystemConfiguration().getCount(
- Property.TSERV_MINC_MAXCONCURRENT), Constants.TSERV_MINC_MAXCONCURRENT_NUMWAITING_MULTIPLIER);
- }
-
@Override
public MemoryManagementActions getMemoryManagementActions(List<TabletState> tablets) {
+ if (maxMemory < 0)
+ throw new IllegalStateException("need to initialize Largst");
mincIdleThresholds.clear();
long ingestMemory = 0;
long compactionMemory = 0;
@@ -92,7 +101,7 @@ public class LargestFirstMemoryManager i
}
Text tableId = ts.getExtent().getTableId();
if (!mincIdleThresholds.containsKey(tableId))
- mincIdleThresholds.put(tableId, ServerConfiguration.getTableConfiguration(tableId.toString()).getTimeInMillis(Property.TABLE_MINC_COMPACT_IDLETIME));
+ mincIdleThresholds.put(tableId, config.getTableConfiguration(tableId.toString()).getTimeInMillis(Property.TABLE_MINC_COMPACT_IDLETIME));
if (idleTime > mincIdleThresholds.get(tableId) && tml > largestIdleMemTableLoad) {
largestIdleMemTableLoad = tml;
largestIdleMemTablet = ts.getExtent();
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/MemoryManager.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/MemoryManager.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/MemoryManager.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/MemoryManager.java Fri Apr 6 16:35:29 2012
@@ -19,6 +19,7 @@ package org.apache.accumulo.server.table
import java.util.List;
import org.apache.accumulo.core.data.KeyExtent;
+import org.apache.accumulo.server.conf.ServerConfiguration;
/**
* A MemoryManager in accumulo currently determines when minor compactions should occur and when ingest should be put on hold. The goal of a memory manager
@@ -29,6 +30,14 @@ import org.apache.accumulo.core.data.Key
*/
public interface MemoryManager {
+
+ /**
+ * Initialize the memory manager.
+ *
+ * @param conf
+ */
+ void init(ServerConfiguration conf);
+
/**
* An implementation of this function will be called periodically by accumulo and should return a list of tablets to minor compact.
*
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/Tablet.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/Tablet.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/Tablet.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/Tablet.java Fri Apr 6 16:35:29 2012
@@ -51,6 +51,7 @@ import org.apache.accumulo.cloudtrace.in
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.impl.ScannerImpl;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.ConfigurationObserver;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.constraints.Violations;
@@ -240,7 +241,7 @@ public class Tablet {
private CommitSession commitSession;
TabletMemory() {
- memTable = new InMemoryMap();
+ memTable = new InMemoryMap(tabletServer.getSystemConfiguration());
commitSession = new CommitSession(nextSeq, memTable);
nextSeq += 2;
}
@@ -263,7 +264,7 @@ public class Tablet {
}
otherMemTable = memTable;
- memTable = new InMemoryMap();
+ memTable = new InMemoryMap(tabletServer.getSystemConfiguration());
CommitSession oldCommitSession = commitSession;
commitSession = new CommitSession(nextSeq, memTable);
@@ -454,7 +455,7 @@ public class Tablet {
}
String getNextMapFilename(String prefix) throws IOException {
- String extension = FileOperations.getNewFileExtension(ServerConfiguration.getTableConfiguration(extent.getTableId().toString()));
+ String extension = FileOperations.getNewFileExtension(tabletServer.getTableConfiguration(extent));
checkTabletDir();
return location.toString() + "/" + prefix + UniqueNameAllocator.getInstance().getNextName() + "." + extension;
}
@@ -480,10 +481,6 @@ public class Tablet {
}
}
- public boolean isMetadataTablet() {
- return extent.getTableId().toString().equals(Constants.METADATA_TABLE_ID);
- }
-
private static String rel2abs(String relPath, KeyExtent extent) {
if (relPath.startsWith("../"))
return ServerConstants.getTablesDir() + relPath.substring(2);
@@ -658,7 +655,7 @@ public class Tablet {
}
- if (extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (extent.isRootTablet()) {
throw new IllegalArgumentException("Can not import files to root tablet");
}
@@ -724,7 +721,7 @@ public class Tablet {
if (mergingMinorCompactionFile != null)
throw new IllegalStateException("Tried to reserve merging minor compaction file when already reserved : " + mergingMinorCompactionFile);
- if (extent.equals(Constants.ROOT_TABLET_EXTENT))
+ if (extent.isRootTablet())
return null;
int maxFiles = acuTableConf.getMaxFilesPerTablet();
@@ -777,7 +774,7 @@ public class Tablet {
void bringMinorCompactionOnline(Path tmpDatafile, Path newDatafile, Path absMergeFile, DataFileValue dfv, CommitSession commitSession, long flushId) {
IZooReaderWriter zoo = ZooReaderWriter.getRetryingInstance();
- if (extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (extent.isRootTablet()) {
try {
if (!zoo.isLockHeld(tabletServer.getLock().getLockID())) {
throw new IllegalStateException();
@@ -963,7 +960,7 @@ public class Tablet {
void bringMajorCompactionOnline(Set<Path> oldDatafiles, Path tmpDatafile, Path newDatafile, Long compactionId, DataFileValue dfv) throws IOException {
long t1, t2;
- if (!extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (!extent.isRootTablet()) {
if (fs.exists(newDatafile)) {
log.error("Target map file already exist " + newDatafile, new Exception());
@@ -989,7 +986,7 @@ public class Tablet {
dataSourceDeletions.incrementAndGet();
- if (extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (extent.isRootTablet()) {
waitForScansToFinish(oldDatafiles, true, Long.MAX_VALUE);
@@ -1058,7 +1055,7 @@ public class Tablet {
t2 = System.currentTimeMillis();
}
- if (!extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (!extent.isRootTablet()) {
Set<Path> filesInUseByScans = waitForScansToFinish(oldDatafiles, false, 10000);
if (filesInUseByScans.size() > 0)
log.debug("Adding scan refs to metadata " + extent + " " + abs2rel(filesInUseByScans));
@@ -1135,15 +1132,15 @@ public class Tablet {
datafiles, time, null, new HashSet<String>(), initFlushID, initCompactID);
}
- private static String lookupTime(KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
+ private static String lookupTime(AccumuloConfiguration conf, KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
SortedMap<Key,Value> entries;
- if (extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (extent.isRootTablet()) {
return null;
- } else if (extent.getTableId().toString().equals(Constants.METADATA_TABLE_ID)) {
+ } else if (extent.isMeta()) {
SortedSet<Column> columns = new TreeSet<Column>();
columns.add(Constants.METADATA_TIME_COLUMN.toColumn());
- entries = MetadataTable.getRootMetadataDataEntries(extent, columns, SecurityConstants.getSystemCredentials());
+ entries = MetadataTable.getRootMetadataDataEntries(conf, extent, columns, SecurityConstants.getSystemCredentials());
} else {
entries = new TreeMap<Key,Value>();
Text rowName = extent.getMetadataEntry();
@@ -1161,13 +1158,14 @@ public class Tablet {
return null;
}
- private static SortedMap<String,DataFileValue> lookupDatafiles(Text locText, FileSystem fs, KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues)
+ private static SortedMap<String,DataFileValue> lookupDatafiles(AccumuloConfiguration conf, Text locText, FileSystem fs, KeyExtent extent,
+ SortedMap<Key,Value> tabletsKeyValues)
throws IOException {
Path location = new Path(ServerConstants.getTablesDir() + "/" + extent.getTableId().toString() + locText.toString());
TreeMap<String,DataFileValue> datafiles = new TreeMap<String,DataFileValue>();
- if (extent.equals(Constants.ROOT_TABLET_EXTENT)) { // the meta0 tablet
+ if (extent.isRootTablet()) { // the meta0 tablet
// cleanUpFiles() has special handling for delete. files
FileStatus[] files = fs.listStatus(location);
Path[] paths = new Path[files.length];
@@ -1185,8 +1183,8 @@ public class Tablet {
SortedMap<Key,Value> datafilesMetadata;
- if (extent.getTableId().toString().equals(Constants.METADATA_TABLE_ID)) {
- datafilesMetadata = MetadataTable.getRootMetadataDataFileEntries(extent, SecurityConstants.getSystemCredentials());
+ if (extent.isMeta()) {
+ datafilesMetadata = MetadataTable.getRootMetadataDataFileEntries(conf, extent, SecurityConstants.getSystemCredentials());
} else {
Text rowName = extent.getMetadataEntry();
@@ -1252,7 +1250,7 @@ public class Tablet {
private static List<LogEntry> lookupLogEntries(KeyExtent ke, SortedMap<Key,Value> tabletsKeyValues) {
List<LogEntry> logEntries = new ArrayList<LogEntry>();
- if (ke.getTableId().toString().equals(Constants.METADATA_TABLE_ID)) {
+ if (ke.isMeta()) {
try {
logEntries = MetadataTable.getLogEntries(SecurityConstants.getSystemCredentials(), ke);
} catch (Exception ex) {
@@ -1313,9 +1311,10 @@ public class Tablet {
private Tablet(TabletServer tabletServer, Text location, KeyExtent extent, TabletResourceManager trm, Configuration conf, FileSystem fs,
SortedMap<Key,Value> tabletsKeyValues) throws IOException {
- this(tabletServer, location, extent, trm, conf, fs, lookupLogEntries(extent, tabletsKeyValues), lookupDatafiles(location, fs, extent, tabletsKeyValues),
- lookupTime(extent, tabletsKeyValues), lookupLastServer(extent, tabletsKeyValues), lookupScanFiles(extent, tabletsKeyValues), lookupFlushID(extent,
- tabletsKeyValues), lookupCompactID(extent, tabletsKeyValues));
+ this(tabletServer, location, extent, trm, conf, fs, lookupLogEntries(extent, tabletsKeyValues), lookupDatafiles(tabletServer.getSystemConfiguration(),
+ location, fs, extent, tabletsKeyValues),
+ lookupTime(tabletServer.getSystemConfiguration(), extent, tabletsKeyValues), lookupLastServer(extent,
+ tabletsKeyValues), lookupScanFiles(extent, tabletsKeyValues), lookupFlushID(extent, tabletsKeyValues), lookupCompactID(extent, tabletsKeyValues));
}
private static TServerInstance lookupLastServer(KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
@@ -1337,7 +1336,7 @@ public class Tablet {
this.lastLocation = lastLocation;
this.tabletDirectory = location.toString();
this.conf = conf;
- this.acuTableConf = ServerConfiguration.getTableConfiguration(extent.getTableId().toString());
+ this.acuTableConf = tabletServer.getTableConfiguration(extent);
this.fs = fs;
this.extent = extent;
@@ -1346,14 +1345,14 @@ public class Tablet {
this.lastFlushID = initFlushID;
this.lastCompactID = initCompactID;
- if (extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (extent.isRootTablet()) {
long rtime = Long.MIN_VALUE;
for (String path : datafiles.keySet()) {
String filename = new Path(path).getName();
FileSKVIterator reader = FileOperations.getInstance().openReader(this.location + "/" + filename, true, fs, fs.getConf(),
- ServerConfiguration.getTableConfiguration(Constants.METADATA_TABLE_ID));
+ tabletServer.getTableConfiguration(extent));
long maxTime = -1;
try {
@@ -1392,7 +1391,7 @@ public class Tablet {
try {
log.debug("Reloading constraints");
- cc = ConstraintLoader.load(extent.getTableId().toString());
+ cc = ConstraintLoader.load(getTableConfiguration());
} catch (IOException e) {
log.error("Failed to reload constraints for " + extent, e);
cc = new ConstraintChecker();
@@ -1498,7 +1497,7 @@ public class Tablet {
}
private void setupDefaultSecurityLabels(KeyExtent extent) {
- if (extent.getTableId().toString().equals(Constants.METADATA_TABLE_ID)) {
+ if (extent.isMeta()) {
defaultSecurityLabel = new byte[0];
} else {
try {
@@ -2740,7 +2739,7 @@ public class Tablet {
throw new RuntimeException(msg);
}
- if (extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (extent.isRootTablet()) {
if (!fileLog.getSecond().keySet().equals(datafileManager.getDatafileSizesRel().keySet())) {
String msg = "Data file in !METADATA differ from in memory data " + extent + " " + fileLog.getSecond().keySet() + " "
+ datafileManager.getDatafileSizesRel().keySet();
@@ -3013,7 +3012,7 @@ public class Tablet {
try {
// we should make .25 below configurable
- keys = FileUtil.findMidPoint(fs, ServerConfiguration.getSystemConfiguration(), extent.getPrevEndRow(), extent.getEndRow(), files, .25);
+ keys = FileUtil.findMidPoint(fs, tabletServer.getSystemConfiguration(), extent.getPrevEndRow(), extent.getEndRow(), files, .25);
} catch (IOException e) {
log.error("Failed to find midpoint " + e.getMessage());
return null;
@@ -3024,7 +3023,7 @@ public class Tablet {
Text lastRow;
if (extent.getEndRow() == null) {
- Key lastKey = (Key) FileUtil.findLastKey(fs, ServerConfiguration.getSystemConfiguration(), files);
+ Key lastKey = (Key) FileUtil.findLastKey(fs, tabletServer.getSystemConfiguration(), files);
lastRow = lastKey.getRow();
} else {
lastRow = extent.getEndRow();
@@ -3130,7 +3129,7 @@ public class Tablet {
majorCompactionWaitingToStart = false;
notifyAll();
- if (extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (extent.isRootTablet()) {
// very important that we call this before doing major compaction,
// otherwise deleted compacted files could possible be brought back
// at some point if the file they were compacted to was legitimately
@@ -3427,7 +3426,7 @@ public class Tablet {
throw new IllegalArgumentException();
}
- if (extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (extent.isRootTablet()) {
String msg = "Cannot split root tablet";
log.warn(msg);
throw new RuntimeException(msg);
@@ -3446,7 +3445,7 @@ public class Tablet {
// from the set... can still query and insert into the tablet while this
// map file operation is happening
Map<String,org.apache.accumulo.core.file.FileUtil.FileInfo> firstAndLastRowsAbs = FileUtil.tryToGetFirstAndLastRows(fs,
- ServerConfiguration.getSystemConfiguration(), datafileManager.getFiles());
+ tabletServer.getSystemConfiguration(), datafileManager.getFiles());
// convert absolute paths to relative paths
Map<String,org.apache.accumulo.core.file.FileUtil.FileInfo> firstAndLastRows = new HashMap<String,org.apache.accumulo.core.file.FileUtil.FileInfo>();
@@ -3467,7 +3466,7 @@ public class Tablet {
splitPoint = findSplitRow(datafileManager.getFiles());
else {
Text tsp = new Text(sp);
- splitPoint = new SplitRowSpec(FileUtil.estimatePercentageLTE(fs, ServerConfiguration.getSystemConfiguration(), extent.getPrevEndRow(),
+ splitPoint = new SplitRowSpec(FileUtil.estimatePercentageLTE(fs, tabletServer.getSystemConfiguration(), extent.getPrevEndRow(),
extent.getEndRow(), datafileManager.getFiles(), tsp), tsp);
}
@@ -3584,7 +3583,7 @@ public class Tablet {
// TODO check seems uneeded now
long lockWait = System.currentTimeMillis() - now;
- if (lockWait > ServerConfiguration.getSystemConfiguration().getTimeInMillis(Property.GENERAL_RPC_TIMEOUT)) {
+ if (lockWait > tabletServer.getSystemConfiguration().getTimeInMillis(Property.GENERAL_RPC_TIMEOUT)) {
throw new IOException("Timeout waiting " + (lockWait / 1000.) + " seconds to get tablet lock");
}
@@ -3783,4 +3782,7 @@ public class Tablet {
}
}
+ public TableConfiguration getTableConfiguration() {
+ return tabletServer.getTableConfiguration(extent);
+ }
}
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java Fri Apr 6 16:35:29 2012
@@ -135,6 +135,7 @@ import org.apache.accumulo.server.Server
import org.apache.accumulo.server.client.ClientServiceHandler;
import org.apache.accumulo.server.client.HdfsZooInstance;
import org.apache.accumulo.server.conf.ServerConfiguration;
+import org.apache.accumulo.server.conf.TableConfiguration;
import org.apache.accumulo.server.master.state.Assignment;
import org.apache.accumulo.server.master.state.DistributedStoreException;
import org.apache.accumulo.server.master.state.TServerInstance;
@@ -220,9 +221,12 @@ public class TabletServer extends Abstra
protected TabletServerMinCMetrics mincMetrics = new TabletServerMinCMetrics();
- public TabletServer(Instance instance, FileSystem fs) {
+ private ServerConfiguration serverConfig;
+
+ public TabletServer(ServerConfiguration conf, FileSystem fs) {
super();
- this.instance = instance;
+ this.serverConfig = conf;
+ this.instance = conf.getInstance();
this.fs = TraceFileSystem.wrap(fs);
SimpleTimer.getInstance().schedule(new TimerTask() {
@Override
@@ -240,7 +244,7 @@ public class TabletServer extends Abstra
}, 5000, 5000);
}
- private synchronized static void logGCInfo() {
+ private synchronized static void logGCInfo(AccumuloConfiguration conf) {
List<GarbageCollectorMXBean> gcmBeans = ManagementFactory.getGarbageCollectorMXBeans();
Runtime rt = Runtime.getRuntime();
@@ -295,7 +299,7 @@ public class TabletServer extends Abstra
log.debug(sb.toString());
}
- final long keepAliveTimeout = ServerConfiguration.getSystemConfiguration().getTimeInMillis(Property.INSTANCE_ZK_TIMEOUT);
+ final long keepAliveTimeout = conf.getTimeInMillis(Property.INSTANCE_ZK_TIMEOUT);
if (maxIncreaseInCollectionTime > keepAliveTimeout) {
Halt.halt("Garbage collection may be interfering with lock keep-alive. Halting.", -1);
}
@@ -320,11 +324,11 @@ public class TabletServer extends Abstra
SecureRandom random;
Map<Long,Session> sessions;
- SessionManager() {
+ SessionManager(AccumuloConfiguration conf) {
random = new SecureRandom();
sessions = new HashMap<Long,Session>();
- final long maxIdle = ServerConfiguration.getSystemConfiguration().getTimeInMillis(Property.TSERV_SESSION_MAXIDLE);
+ final long maxIdle = conf.getTimeInMillis(Property.TSERV_SESSION_MAXIDLE);
TimerTask r = new TimerTask() {
public void run() {
@@ -811,13 +815,17 @@ public class TabletServer extends Abstra
}
}
+ public AccumuloConfiguration getSystemConfiguration() {
+ return serverConfig.getConfiguration();
+ }
+
TransactionWatcher watcher = new TransactionWatcher();
private class ThriftClientHandler extends ClientServiceHandler implements TabletClientService.Iface {
SessionManager sessionManager;
- AccumuloConfiguration acuConf = ServerConfiguration.getSystemConfiguration();
+ AccumuloConfiguration acuConf = getSystemConfiguration();
TabletServerUpdateMetrics updateMetrics = new TabletServerUpdateMetrics();
@@ -826,9 +834,9 @@ public class TabletServer extends Abstra
WriteTracker writeTracker = new WriteTracker();
ThriftClientHandler() {
- super(watcher);
+ super(instance, watcher);
log.debug(ThriftClientHandler.class.getName() + " created");
- sessionManager = new SessionManager();
+ sessionManager = new SessionManager(getSystemConfiguration());
// Register the metrics MBean
try {
updateMetrics.register();
@@ -1237,10 +1245,9 @@ public class TabletServer extends Abstra
for (KeyExtent keyExtent : batch.keySet()) {
if (threadPoolExtent == null) {
threadPoolExtent = keyExtent;
- } else if (keyExtent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ } else if (keyExtent.isRootTablet()) {
throw new IllegalArgumentException("Cannot batch query root tablet with other tablets " + threadPoolExtent + " " + keyExtent);
- } else if (keyExtent.getTableId().toString().equals(Constants.METADATA_TABLE_ID)
- && !threadPoolExtent.getTableId().toString().equals(Constants.METADATA_TABLE_ID)) {
+ } else if (keyExtent.isMeta() && !threadPoolExtent.isMeta()) {
throw new IllegalArgumentException("Cannot batch query !METADATA and non !METADATA tablets " + threadPoolExtent + " " + keyExtent);
}
@@ -1428,7 +1435,7 @@ public class TabletServer extends Abstra
mutations.add(mutation);
us.queuedMutationSize += mutation.numBytes();
}
- if (us.queuedMutationSize > ServerConfiguration.getSystemConfiguration().getMemoryInBytes(Property.TSERV_MUTATION_QUEUE_MAX))
+ if (us.queuedMutationSize > getSystemConfiguration().getMemoryInBytes(Property.TSERV_MUTATION_QUEUE_MAX))
flush(us);
}
} finally {
@@ -1446,7 +1453,7 @@ public class TabletServer extends Abstra
boolean containsMetadataTablet = false;
for (Tablet tablet : us.queuedMutations.keySet())
- if (tablet.getExtent().getTableId().toString().equals(Constants.METADATA_TABLE_ID))
+ if (tablet.getExtent().isMeta())
containsMetadataTablet = true;
if (!containsMetadataTablet && us.queuedMutations.size() > 0)
@@ -1633,7 +1640,7 @@ public class TabletServer extends Abstra
throw new NotServingTabletException(tkeyExtent);
}
- if (!keyExtent.getTableId().toString().equals(Constants.METADATA_TABLE_ID))
+ if (!keyExtent.isMeta())
TabletServer.this.resourceManager.waitUntilCommitsAreEnabled();
long opid = writeTracker.startWrite(TabletType.type(keyExtent));
@@ -1754,7 +1761,7 @@ public class TabletServer extends Abstra
if (fatal) {
Halt.halt(1, new Runnable() {
public void run() {
- logGCInfo();
+ logGCInfo(getSystemConfiguration());
}
});
}
@@ -1770,7 +1777,7 @@ public class TabletServer extends Abstra
Halt.halt(1, new Runnable() {
public void run() {
log.info("Tablet server no longer holds lock during checkPermission() : " + request + ", exiting");
- logGCInfo();
+ logGCInfo(getSystemConfiguration());
}
});
}
@@ -1836,7 +1843,7 @@ public class TabletServer extends Abstra
final Runnable ah = new LoggingRunnable(log, new AssignmentHandler(extent));
// Root tablet assignment must take place immediately
- if (extent.compareTo(Constants.ROOT_TABLET_EXTENT) == 0) {
+ if (extent.isRootTablet()) {
new Thread("Root Tablet Assignment") {
public void run() {
ah.run();
@@ -1849,7 +1856,7 @@ public class TabletServer extends Abstra
}
}.start();
} else {
- if (extent.getTableId().compareTo(new Text(Constants.METADATA_TABLE_ID)) == 0) {
+ if (extent.isMeta()) {
resourceManager.addMetaDataAssignment(ah);
} else {
resourceManager.addAssignment(ah);
@@ -1935,7 +1942,7 @@ public class TabletServer extends Abstra
Halt.halt(0, new Runnable() {
public void run() {
log.info("Master requested tablet server halt");
- logGCInfo();
+ logGCInfo(getSystemConfiguration());
serverStopRequested = true;
try {
tabletServerLock.unlock();
@@ -2055,12 +2062,11 @@ public class TabletServer extends Abstra
}
private class MajorCompactor implements Runnable {
- private AccumuloConfiguration acuConf = ServerConfiguration.getSystemConfiguration();
public void run() {
while (!majorCompactorDisabled) {
try {
- UtilWaitThread.sleep(acuConf.getTimeInMillis(Property.TSERV_MAJC_DELAY));
+ UtilWaitThread.sleep(getSystemConfiguration().getTimeInMillis(Property.TSERV_MAJC_DELAY));
TreeMap<KeyExtent,Tablet> copyOnlineTablets = new TreeMap<KeyExtent,Tablet>();
@@ -2092,8 +2098,7 @@ public class TabletServer extends Abstra
continue;
}
- int maxLogEntriesPerTablet = ServerConfiguration.getTableConfiguration(tablet.getExtent().getTableId().toString()).getCount(
- Property.TABLE_MINC_LOGS_MAX);
+ int maxLogEntriesPerTablet = getTableConfiguration(tablet.getExtent()).getCount(Property.TABLE_MINC_LOGS_MAX);
if (tablet.getLogCount() >= maxLogEntriesPerTablet) {
log.debug("Initiating minor compaction for " + tablet.getExtent() + " because it has " + tablet.getLogCount() + " write ahead logs");
@@ -2108,7 +2113,7 @@ public class TabletServer extends Abstra
}
}
- int idleCompactionsToStart = Math.max(1, acuConf.getCount(Property.TSERV_MAJC_MAXCONCURRENT) / 2);
+ int idleCompactionsToStart = Math.max(1, getSystemConfiguration().getCount(Property.TSERV_MAJC_MAXCONCURRENT) / 2);
if (numMajorCompactionsInProgress < idleCompactionsToStart) {
// system is not major compacting, can schedule some
@@ -2308,8 +2313,6 @@ public class TabletServer extends Abstra
public void run() {
log.info(clientAddress + ": got assignment from master: " + extent);
- final boolean isMetaDataTablet = extent.getTableId().toString().compareTo(Constants.METADATA_TABLE_ID) == 0;
-
synchronized (unopenedTablets) {
synchronized (openingTablets) {
synchronized (onlineTablets) {
@@ -2461,8 +2464,8 @@ public class TabletServer extends Abstra
public void run() {
log.info("adding tablet " + extent + " back to the assignment pool (retry " + retryAttempt + ")");
AssignmentHandler handler = new AssignmentHandler(extentToOpen, retryAttempt + 1);
- if (isMetaDataTablet) {
- if (Constants.ROOT_TABLET_EXTENT.equals(extent)) {
+ if (extent.isMeta()) {
+ if (extent.isRootTablet()) {
new Thread(new LoggingRunnable(log, handler), "Root tablet assignment retry").start();
} else {
resourceManager.addMetaDataAssignment(handler);
@@ -2554,8 +2557,8 @@ public class TabletServer extends Abstra
MetadataTable.addLogEntries(SecurityConstants.getSystemCredentials(), entries, getLock());
}
- private int startServer(Property portHint, TProcessor processor, String threadName) throws UnknownHostException {
- ServerPort sp = TServerUtils.startServer(portHint, processor, this.getClass().getSimpleName(), threadName, Property.TSERV_PORTSEARCH,
+ private int startServer(AccumuloConfiguration conf, Property portHint, TProcessor processor, String threadName) throws UnknownHostException {
+ ServerPort sp = TServerUtils.startServer(conf, portHint, processor, this.getClass().getSimpleName(), threadName, Property.TSERV_PORTSEARCH,
Property.TSERV_MINTHREADS, Property.TSERV_THREADCHECK);
this.server = sp.server;
return sp.port;
@@ -2581,7 +2584,7 @@ public class TabletServer extends Abstra
return null;
}
MasterClientService.Iface client = ThriftUtil.getClient(new MasterClientService.Client.Factory(), address, Property.MASTER_CLIENTPORT,
- Property.GENERAL_RPC_TIMEOUT, ServerConfiguration.getSystemConfiguration());
+ Property.GENERAL_RPC_TIMEOUT, getSystemConfiguration());
// log.info("Listener API to master has been opened");
return client;
} catch (Exception e) {
@@ -2598,7 +2601,7 @@ public class TabletServer extends Abstra
// start listening for client connection last
TabletClientService.Iface tch = TraceWrap.service(new ThriftClientHandler());
TabletClientService.Processor processor = new TabletClientService.Processor(tch);
- int port = startServer(Property.TSERV_CLIENTPORT, processor, "Thrift Client Server");
+ int port = startServer(getSystemConfiguration(), Property.TSERV_CLIENTPORT, processor, "Thrift Client Server");
log.info("port = " + port);
return port;
}
@@ -2624,7 +2627,7 @@ public class TabletServer extends Abstra
public void run() {
if (!serverStopRequested)
log.fatal("Lost tablet server lock (reason = " + reason + "), exiting.");
- logGCInfo();
+ logGCInfo(getSystemConfiguration());
}
});
}
@@ -2763,7 +2766,7 @@ public class TabletServer extends Abstra
log.warn("Failed to close filesystem : " + e.getMessage(), e);
}
- logGCInfo();
+ logGCInfo(getSystemConfiguration());
log.info("TServerInfo: stop requested. exiting ... ");
@@ -2781,7 +2784,7 @@ public class TabletServer extends Abstra
for (int tries = 0; tries < 3; tries++) {
try {
log.debug("verifying extent " + extent);
- if (extent.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (extent.isRootTablet()) {
TreeMap<KeyExtent,Text> set = new TreeMap<KeyExtent,Text>();
set.put(extent, new Text(Constants.ZROOT_TABLET));
return set;
@@ -2934,9 +2937,9 @@ public class TabletServer extends Abstra
log.info("Tablet server starting on " + hostname);
authenticator = ZKAuthenticator.getInstance();
clientAddress = new InetSocketAddress(hostname, 0);
- logger = new TabletServerLogger(this, ServerConfiguration.getSystemConfiguration().getMemoryInBytes(Property.TSERV_WALOG_MAX_SIZE));
+ logger = new TabletServerLogger(this, getSystemConfiguration().getMemoryInBytes(Property.TSERV_WALOG_MAX_SIZE));
- if (ServerConfiguration.getSystemConfiguration().getBoolean(Property.TSERV_LOCK_MEMORY)) {
+ if (getSystemConfiguration().getBoolean(Property.TSERV_LOCK_MEMORY)) {
String path = "lib/native/mlock/" + System.mapLibraryName("MLock-" + Platform.getPlatform());
path = new File(path).getAbsolutePath();
try {
@@ -2951,18 +2954,18 @@ public class TabletServer extends Abstra
}
}
- FileSystemMonitor.start(Property.TSERV_MONITOR_FS);
+ FileSystemMonitor.start(getSystemConfiguration(), Property.TSERV_MONITOR_FS);
TimerTask gcDebugTask = new TimerTask() {
@Override
public void run() {
- logGCInfo();
+ logGCInfo(getSystemConfiguration());
}
};
SimpleTimer.getInstance().schedule(gcDebugTask, 0, 1000);
- this.resourceManager = new TabletServerResourceManager(fs);
+ this.resourceManager = new TabletServerResourceManager(instance, fs);
lastPingTime = System.currentTimeMillis();
@@ -2975,7 +2978,7 @@ public class TabletServer extends Abstra
majorCompactorThread.setName("Split/MajC initiator");
majorCompactorThread.start();
- String className = ServerConfiguration.getSystemConfiguration().get(Property.TSERV_LOGGER_STRATEGY);
+ String className = getSystemConfiguration().get(Property.TSERV_LOGGER_STRATEGY);
Class<? extends LoggerStrategy> klass = DEFAULT_LOGGER_STRATEGY;
try {
klass = AccumuloClassLoader.loadClass(className, LoggerStrategy.class);
@@ -2985,6 +2988,7 @@ public class TabletServer extends Abstra
try {
Constructor<? extends LoggerStrategy> constructor = klass.getConstructor(TabletServer.class);
loggerStrategy = constructor.newInstance(this);
+ loggerStrategy.init(serverConfig);
} catch (Exception ex) {
log.warn("Unable to create object of type " + klass.getName() + " using " + DEFAULT_LOGGER_STRATEGY.getName());
}
@@ -3088,10 +3092,11 @@ public class TabletServer extends Abstra
public static void main(String[] args) throws IOException {
try {
FileSystem fs = FileUtil.getFileSystem(CachedConfiguration.getInstance(), ServerConfiguration.getSiteConfiguration());
- Accumulo.init(fs, "tserver");
String hostname = Accumulo.getLocalAddress(args);
Instance instance = HdfsZooInstance.getInstance();
- TabletServer server = new TabletServer(instance, fs);
+ ServerConfiguration conf = new ServerConfiguration(instance);
+ Accumulo.init(fs, conf, "tserver");
+ TabletServer server = new TabletServer(conf, fs);
server.config(hostname);
Accumulo.enableTracing(hostname, "tserver");
server.run();
@@ -3140,8 +3145,7 @@ public class TabletServer extends Abstra
private final AtomicInteger logIdGenerator = new AtomicInteger();
public int createLogId(KeyExtent tablet) {
- String table = tablet.getTableId().toString();
- AccumuloConfiguration acuTableConf = ServerConfiguration.getTableConfiguration(table);
+ AccumuloConfiguration acuTableConf = getTableConfiguration(tablet);
if (acuTableConf.getBoolean(Property.TABLE_WALOG_ENABLED)) {
return logIdGenerator.incrementAndGet();
}
@@ -3316,4 +3320,12 @@ public class TabletServer extends Abstra
return METRICS_PREFIX;
}
+ // public AccumuloConfiguration getTableConfiguration(String tableId) {
+ // return ServerConfiguration.getTableConfiguration(instance, tableId);
+ // }
+
+ public TableConfiguration getTableConfiguration(KeyExtent extent) {
+ return ServerConfiguration.getTableConfiguration(instance, extent.getTableId().toString());
+ }
+
}
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServerResourceManager.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServerResourceManager.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServerResourceManager.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServerResourceManager.java Fri Apr 6 16:35:29 2012
@@ -37,6 +37,7 @@ import java.util.concurrent.atomic.Atomi
import org.apache.accumulo.cloudtrace.instrument.TraceExecutorService;
import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.KeyExtent;
@@ -76,18 +77,17 @@ public class TabletServerResourceManager
private ExecutorService defaultReadAheadThreadPool;
private Map<String,ExecutorService> threadPools = new TreeMap<String,ExecutorService>();
- private AccumuloConfiguration acuConf;
-
private HashSet<TabletResourceManager> tabletResources;
private FileManager fileManager;
- private MemoryManager memoryManger;
+ private MemoryManager memoryManager;
private MemoryManagementFramework memMgmt;
- private LruBlockCache _dCache = null;
- private LruBlockCache _iCache = null;
+ private final LruBlockCache _dCache;
+ private final LruBlockCache _iCache;
+ private final ServerConfiguration conf;
private static final Logger log = Logger.getLogger(TabletServerResourceManager.class);
@@ -114,12 +114,12 @@ public class TabletServerResourceManager
return addEs(name, new ThreadPoolExecutor(min, max, timeout, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new NamingThreadFactory(name)));
}
- public TabletServerResourceManager(FileSystem fs) {
-
- this.acuConf = ServerConfiguration.getSystemConfiguration();
+ public TabletServerResourceManager(Instance instance, FileSystem fs) {
+ this.conf = new ServerConfiguration(instance);
+ AccumuloConfiguration acuConf = conf.getConfiguration();
long maxMemory = acuConf.getMemoryInBytes(Property.TSERV_MAXMEM);
- boolean usingNativeMap = ServerConfiguration.getSystemConfiguration().getBoolean(Property.TSERV_NATIVEMAP_ENABLED) && NativeMap.loadedNativeLibraries();
+ boolean usingNativeMap = acuConf.getBoolean(Property.TSERV_NATIVEMAP_ENABLED) && NativeMap.loadedNativeLibraries();
long blockSize = acuConf.getMemoryInBytes(Property.TSERV_DEFAULT_BLOCKSIZE);
long dCacheSize = acuConf.getMemoryInBytes(Property.TSERV_DATACACHE_SIZE);
@@ -170,19 +170,20 @@ public class TabletServerResourceManager
int maxOpenFiles = acuConf.getCount(Property.TSERV_SCAN_MAX_OPENFILES);
- fileManager = new FileManager(fs, maxOpenFiles, _dCache, _iCache);
+ fileManager = new FileManager(conf, fs, maxOpenFiles, _dCache, _iCache);
try {
- Class<? extends MemoryManager> clazz = AccumuloClassLoader.loadClass(ServerConfiguration.getSystemConfiguration().get(Property.TSERV_MEM_MGMT),
+ Class<? extends MemoryManager> clazz = AccumuloClassLoader.loadClass(acuConf.get(Property.TSERV_MEM_MGMT),
MemoryManager.class);
- memoryManger = clazz.newInstance();
- log.debug("Loaded memory manager : " + memoryManger.getClass().getName());
+ memoryManager = clazz.newInstance();
+ memoryManager.init(conf);
+ log.debug("Loaded memory manager : " + memoryManager.getClass().getName());
} catch (Exception e) {
log.error("Failed to find memory manger in config, using default", e);
}
- if (memoryManger == null) {
- memoryManger = new LargestFirstMemoryManager();
+ if (memoryManager == null) {
+ memoryManager = new LargestFirstMemoryManager();
}
memMgmt = new MemoryManagementFramework();
@@ -233,7 +234,7 @@ public class TabletServerResourceManager
MemoryManagementFramework() {
tabletReports = Collections.synchronizedMap(new HashMap<KeyExtent,TabletStateImpl>());
memUsageReports = new LinkedBlockingQueue<TabletStateImpl>();
- maxMem = ServerConfiguration.getSystemConfiguration().getMemoryInBytes(Property.TSERV_MAXMEM);
+ maxMem = conf.getConfiguration().getMemoryInBytes(Property.TSERV_MAXMEM);
Runnable r1 = new Runnable() {
public void run() {
@@ -308,7 +309,7 @@ public class TabletServerResourceManager
synchronized (tabletReports) {
tablets = new ArrayList<TabletState>(tabletReports.values());
}
- mma = memoryManger.getMemoryManagementActions(tablets);
+ mma = memoryManager.getMemoryManagementActions(tablets);
} catch (Throwable t) {
log.error("Memory manager failed " + t.getMessage(), t);
@@ -377,7 +378,7 @@ public class TabletServerResourceManager
void waitUntilCommitsAreEnabled() {
if (holdCommits) {
- long timeout = System.currentTimeMillis() + ServerConfiguration.getSystemConfiguration().getTimeInMillis(Property.GENERAL_RPC_TIMEOUT);
+ long timeout = System.currentTimeMillis() + conf.getConfiguration().getTimeInMillis(Property.GENERAL_RPC_TIMEOUT);
synchronized (commitHold) {
while (holdCommits) {
try {
@@ -654,7 +655,7 @@ public class TabletServerResourceManager
TabletServerResourceManager.this.removeTabletResource(this);
memMgmt.tabletClosed(tablet.getExtent());
- memoryManger.tabletClosed(tablet.getExtent());
+ memoryManager.tabletClosed(tablet.getExtent());
closed = true;
}
@@ -672,8 +673,8 @@ public class TabletServerResourceManager
}
public void executeSplit(KeyExtent tablet, Runnable splitTask) {
- if (tablet.getTableId().toString().equals(Constants.METADATA_TABLE_ID)) {
- if (tablet.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (tablet.isMeta()) {
+ if (tablet.isRootTablet()) {
log.warn("Saw request to split root tablet, ignoring");
return;
}
@@ -686,7 +687,7 @@ public class TabletServerResourceManager
public void executeMajorCompaction(KeyExtent tablet, Runnable compactionTask) {
if (tablet.equals(Constants.ROOT_TABLET_EXTENT)) {
rootMajorCompactionThreadPool.execute(compactionTask);
- } else if (tablet.getTableId().toString().equals(Constants.METADATA_TABLE_ID)) {
+ } else if (tablet.isMeta()) {
defaultMajorCompactionThreadPool.execute(compactionTask);
} else {
majorCompactionThreadPool.execute(compactionTask);
@@ -694,9 +695,9 @@ public class TabletServerResourceManager
}
public void executeReadAhead(KeyExtent tablet, Runnable task) {
- if (tablet.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (tablet.isRootTablet()) {
task.run();
- } else if (tablet.getTableId().toString().equals(Constants.METADATA_TABLE_ID)) {
+ } else if (tablet.isMeta()) {
defaultReadAheadThreadPool.execute(task);
} else {
readAheadThreadPool.execute(task);
@@ -712,9 +713,9 @@ public class TabletServerResourceManager
}
public void addMigration(KeyExtent tablet, Runnable migrationHandler) {
- if (tablet.equals(Constants.ROOT_TABLET_EXTENT)) {
+ if (tablet.isRootTablet()) {
migrationHandler.run();
- } else if (tablet.getTableId().toString().equals(Constants.METADATA_TABLE_ID)) {
+ } else if (tablet.isMeta()) {
defaultMigrationPool.execute(migrationHandler);
} else {
migrationPool.execute(migrationHandler);
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/LoggerStrategy.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/LoggerStrategy.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/LoggerStrategy.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/LoggerStrategy.java Fri Apr 6 16:35:29 2012
@@ -22,6 +22,12 @@ import org.apache.accumulo.core.conf.Pro
import org.apache.accumulo.server.conf.ServerConfiguration;
public abstract class LoggerStrategy {
+ private ServerConfiguration configuration = null;
+
+ public void init(ServerConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
// Called by the tablet server to get the list of loggers to use from the available set
public abstract Set<String> getLoggers(Set<String> allLoggers);
@@ -29,6 +35,6 @@ public abstract class LoggerStrategy {
public abstract void preferLoggers(Set<String> preference);
public int getNumberOfLoggersToUse() {
- return ServerConfiguration.getSystemConfiguration().getCount(Property.TSERV_LOGGER_COUNT);
+ return configuration.getConfiguration().getCount(Property.TSERV_LOGGER_COUNT);
}
}
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/RemoteLogger.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/RemoteLogger.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/RemoteLogger.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/RemoteLogger.java Fri Apr 6 16:35:29 2012
@@ -24,6 +24,7 @@ import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Mutation;
@@ -35,7 +36,6 @@ import org.apache.accumulo.core.tabletse
import org.apache.accumulo.core.tabletserver.thrift.TabletMutations;
import org.apache.accumulo.core.util.Daemon;
import org.apache.accumulo.core.util.ThriftUtil;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.server.security.SecurityConstants;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
@@ -162,13 +162,13 @@ public class RemoteLogger {
private final UUID tserverSession;
private MutationLogger.Iface client = null;
- public RemoteLogger(String address, UUID tserverUUID) throws ThriftSecurityException, LoggerClosedException, TException, IOException {
+ public RemoteLogger(String address, UUID tserverUUID, AccumuloConfiguration conf) throws ThriftSecurityException, LoggerClosedException, TException,
+ IOException {
logger = address;
tserverSession = tserverUUID;
try {
- client = ThriftUtil.getClient(new MutationLogger.Client.Factory(), address, Property.LOGGER_PORT, Property.TSERV_LOGGER_TIMEOUT,
- ServerConfiguration.getSystemConfiguration());
+ client = ThriftUtil.getClient(new MutationLogger.Client.Factory(), address, Property.LOGGER_PORT, Property.TSERV_LOGGER_TIMEOUT, conf);
logFile = client.create(null, SecurityConstants.getSystemCredentials(), tserverSession.toString());
log.debug("Got new write-ahead log: " + this);
} catch (ThriftSecurityException tse) {
@@ -190,13 +190,12 @@ public class RemoteLogger {
t.start();
}
- public RemoteLogger(String address) throws IOException {
+ public RemoteLogger(String address, AccumuloConfiguration conf) throws IOException {
logger = address;
tserverSession = null;
logFile = null;
try {
- client = ThriftUtil.getClient(new MutationLogger.Client.Factory(), address, Property.LOGGER_PORT, Property.TSERV_LOGGER_TIMEOUT,
- ServerConfiguration.getSystemConfiguration());
+ client = ThriftUtil.getClient(new MutationLogger.Client.Factory(), address, Property.LOGGER_PORT, Property.TSERV_LOGGER_TIMEOUT, conf);
} catch (TTransportException e) {
throw new IOException(e);
}
@@ -278,8 +277,8 @@ public class RemoteLogger {
client.minorCompactionStarted(null, logFile.id, seq, tid, fqfn);
}
- public synchronized long startCopy(String name, String fullyQualifiedFileName, boolean sort) throws ThriftSecurityException, TException {
- return client.startCopy(null, SecurityConstants.getSystemCredentials(), name, fullyQualifiedFileName, sort);
+ public synchronized long startCopy(String name, String fullyQualifiedFileName) throws ThriftSecurityException, TException {
+ return client.startCopy(null, SecurityConstants.getSystemCredentials(), name, fullyQualifiedFileName, true);
}
public synchronized List<String> getClosedLogs() throws ThriftSecurityException, TException {
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/TabletServerLogger.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/TabletServerLogger.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/TabletServerLogger.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/tabletserver/log/TabletServerLogger.java Fri Apr 6 16:35:29 2012
@@ -38,7 +38,6 @@ import org.apache.accumulo.core.data.thr
import org.apache.accumulo.core.tabletserver.thrift.LoggerClosedException;
import org.apache.accumulo.core.tabletserver.thrift.TabletMutations;
import org.apache.accumulo.core.util.UtilWaitThread;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.server.tabletserver.Tablet;
import org.apache.accumulo.server.tabletserver.Tablet.CommitSession;
import org.apache.accumulo.server.tabletserver.TabletServer;
@@ -79,8 +78,7 @@ public class TabletServerLogger {
private final AtomicInteger seqGen = new AtomicInteger();
private static boolean enabled(Tablet tablet) {
- String table = tablet.getExtent().getTableId().toString();
- return ServerConfiguration.getTableConfiguration(table).getBoolean(Property.TABLE_WALOG_ENABLED);
+ return tablet.getTableConfiguration().getBoolean(Property.TABLE_WALOG_ENABLED);
}
private static boolean enabled(CommitSession commitSession) {
@@ -190,7 +188,7 @@ public class TabletServerLogger {
if (!loggerAddresses.isEmpty()) {
for (String logger : loggerAddresses) {
try {
- loggers.add(new RemoteLogger(logger, UUID.randomUUID()));
+ loggers.add(new RemoteLogger(logger, UUID.randomUUID(), tserver.getSystemConfiguration()));
} catch (LoggerClosedException t) {
close();
break;
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/WrongTabletTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/WrongTabletTest.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/WrongTabletTest.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/WrongTabletTest.java Fri Apr 6 16:35:29 2012
@@ -32,8 +32,9 @@ public class WrongTabletTest {
public static void main(String[] args) {
String location = args[0];
+ ServerConfiguration conf = new ServerConfiguration(HdfsZooInstance.getInstance());
try {
- TabletClientService.Iface client = ThriftUtil.getTServerClient(location, ServerConfiguration.getSystemConfiguration());
+ TabletClientService.Iface client = ThriftUtil.getTServerClient(location, conf.getConfiguration());
Mutation mutation = new Mutation(new Text("row_0003750001"));
// mutation.set(new Text("colf:colq"), new Value("val".getBytes()));
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/BulkFileTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/BulkFileTest.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/BulkFileTest.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/BulkFileTest.java Fri Apr 6 16:35:29 2012
@@ -24,6 +24,7 @@ import java.util.Map.Entry;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.file.FileOperations;
@@ -55,25 +56,24 @@ public class BulkFileTest extends Functi
@Override
public void run() throws Exception {
Configuration conf = new Configuration();
- FileSystem fs = TraceFileSystem.wrap(FileUtil.getFileSystem(conf, ServerConfiguration.getSiteConfiguration()));
+ AccumuloConfiguration aconf = ServerConfiguration.getDefaultConfiguration();
+ FileSystem fs = TraceFileSystem.wrap(FileUtil.getFileSystem(conf, aconf));
String dir = "/tmp/bulk_test_diff_files_89723987592";
fs.delete(new Path(dir), true);
- FileSKVWriter writer1 = FileOperations.getInstance().openWriter(dir + "/f1." + RFile.EXTENSION, fs, conf,
- ServerConfiguration.getSystemConfiguration());
+ FileSKVWriter writer1 = FileOperations.getInstance().openWriter(dir + "/f1." + RFile.EXTENSION, fs, conf, aconf);
writer1.startDefaultLocalityGroup();
writeData(writer1, 0, 333);
writer1.close();
- FileSKVWriter writer2 = FileOperations.getInstance().openWriter(dir + "/f2." + RFile.EXTENSION, fs, conf,
- ServerConfiguration.getSystemConfiguration());
+ FileSKVWriter writer2 = FileOperations.getInstance().openWriter(dir + "/f2." + RFile.EXTENSION, fs, conf, aconf);
writer2.startDefaultLocalityGroup();
writeData(writer2, 334, 999);
writer2.close();
- FileSKVWriter writer3 = FileOperations.getInstance().openWriter(dir + "/f3." + RFile.EXTENSION, fs, conf, ServerConfiguration.getSystemConfiguration());
+ FileSKVWriter writer3 = FileOperations.getInstance().openWriter(dir + "/f3." + RFile.EXTENSION, fs, conf, aconf);
writer3.startDefaultLocalityGroup();
writeData(writer3, 1000, 1999);
writer3.close();
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/FunctionalTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/FunctionalTest.java?rev=1310457&r1=1310456&r2=1310457&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/FunctionalTest.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/FunctionalTest.java Fri Apr 6 16:35:29 2012
@@ -161,7 +161,7 @@ public abstract class FunctionalTest {
}
protected Instance getInstance() {
- return new ZooKeeperInstance(getInstanceName(), ServerConfiguration.getSystemConfiguration().get(Property.INSTANCE_ZK_HOST));
+ return new ZooKeeperInstance(getInstanceName(), ServerConfiguration.getSiteConfiguration().get(Property.INSTANCE_ZK_HOST));
}
protected void setInstanceName(String instanceName) {