Repository: accumulo Updated Branches: refs/heads/master 47a1a72a7 -> dcc19ccba
ACCUMULO-1961 Re-apply inadvertently dropped 4abb3f1 to master branch Fixes trivial warnings and broken javadocs which have been recently introduced. Specifically, removes references to private and package-private (default) classes in public javadoc comments (internal details aren't relevant to the API and subject to change). Another common warning was unused imports and javadoc param tags that refer to non-existent parameters. Commits against the following JIRA issues introduced these: ACCUMULO-1948, ACCUMULO-1974, ACCUMULO-2021, ACCUMULO-2136, ACCUMULO-2322, ACCUMULO-2334, ACCUMULO-2350 Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/dcc19ccb Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/dcc19ccb Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/dcc19ccb Branch: refs/heads/master Commit: dcc19ccbada8c2f0a206ec797455294015e8ca6d Parents: 47a1a72 Author: Christopher Tubbs <ctubb...@apache.org> Authored: Wed Feb 19 14:59:58 2014 -0500 Committer: Christopher Tubbs <ctubb...@apache.org> Committed: Wed Feb 19 14:59:58 2014 -0500 ---------------------------------------------------------------------- .../apache/accumulo/core/client/Instance.java | 36 +++--- .../org/apache/accumulo/core/conf/Property.java | 87 +++++++------- .../accumulo/core/util/ValidatorTest.java | 6 +- .../java/org/apache/accumulo/fate/Fate.java | 67 ++++++----- .../server/watcher/MonitorLog4jWatcher.java | 10 +- .../tserver/log/TabletServerLogger.java | 112 ++++++++++--------- .../tserver/TabletResourceManagerTest.java | 8 +- 7 files changed, 158 insertions(+), 168 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/accumulo/blob/dcc19ccb/core/src/main/java/org/apache/accumulo/core/client/Instance.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/accumulo/core/client/Instance.java b/core/src/main/java/org/apache/accumulo/core/client/Instance.java index c6f6f33..bcdd0e5 100644 --- a/core/src/main/java/org/apache/accumulo/core/client/Instance.java +++ b/core/src/main/java/org/apache/accumulo/core/client/Instance.java @@ -26,54 +26,54 @@ import org.apache.accumulo.core.conf.AccumuloConfiguration; /** * This class represents the information a client needs to know to connect to an instance of accumulo. - * + * */ public interface Instance { /** * Returns the location of the tablet server that is serving the root tablet. - * + * * @return location in "hostname:port" form */ String getRootTabletLocation(); /** * Returns the location(s) of the accumulo master and any redundant servers. - * + * * @return a list of locations in "hostname:port" form */ List<String> getMasterLocations(); /** * Returns a unique string that identifies this instance of accumulo. - * + * * @return a UUID */ String getInstanceID(); /** * Returns the instance name given at system initialization time. - * + * * @return current instance name */ String getInstanceName(); /** * Returns a comma-separated list of zookeeper servers the instance is using. - * + * * @return the zookeeper servers this instance is using in "hostname:port" form */ String getZooKeepers(); /** * Returns the zookeeper connection timeout. - * + * * @return the configured timeout to connect to zookeeper */ int getZooKeepersSessionTimeOut(); /** * Returns a connection to accumulo. - * + * * @param user * a valid accumulo user * @param pass @@ -90,7 +90,7 @@ public interface Instance { /** * Returns a connection to accumulo. - * + * * @param user * a valid accumulo user * @param pass @@ -107,7 +107,7 @@ public interface Instance { /** * Returns a connection to this instance of accumulo. - * + * * @param user * a valid accumulo user * @param pass @@ -121,25 +121,23 @@ public interface Instance { */ @Deprecated Connector getConnector(String user, CharSequence pass) throws AccumuloException, AccumuloSecurityException; - + /** * Returns the AccumuloConfiguration to use when interacting with this instance. - * + * * @return the AccumuloConfiguration that specifies properties related to interacting with this instance - * @deprecated since 1.6.0. - * This method makes very little sense in the context of the client API and never should have been exposed. - * @see {@link InstanceOperations#getConfiguration()} for client-side reading of the server-side configuration. + * @deprecated since 1.6.0. This method makes very little sense in the context of the client API and never should have been exposed. + * @see {@link InstanceOperations#getSystemConfiguration()} for client-side reading of the server-side configuration. */ @Deprecated AccumuloConfiguration getConfiguration(); /** * Set the AccumuloConfiguration to use when interacting with this instance. - * + * * @param conf * accumulo configuration - * @deprecated since 1.6.0. - * This method makes very little sense in the context of the client API and never should have been exposed. + * @deprecated since 1.6.0. This method makes very little sense in the context of the client API and never should have been exposed. * @see {@link InstanceOperations#setProperty(String, String)} */ @Deprecated @@ -147,7 +145,7 @@ public interface Instance { /** * Returns a connection to this instance of accumulo. - * + * * @param principal * a valid accumulo user * @param token http://git-wip-us.apache.org/repos/asf/accumulo/blob/dcc19ccb/core/src/main/java/org/apache/accumulo/core/conf/Property.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java index e260c3a..cc61171 100644 --- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java +++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java @@ -294,8 +294,9 @@ public enum Property { @Sensitive MONITOR_SSL_TRUSTSTOREPASS("monitor.ssl.trustStorePassword", "", PropertyType.STRING, "The truststore password for enabling monitor SSL."), - MONITOR_LOCK_CHECK_INTERVAL("monitor.lock.check.interval", "5s", PropertyType.TIMEDURATION, "The amount of time to sleep between checking for the Montior ZooKeeper lock"), - + MONITOR_LOCK_CHECK_INTERVAL("monitor.lock.check.interval", "5s", PropertyType.TIMEDURATION, + "The amount of time to sleep between checking for the Montior ZooKeeper lock"), + TRACE_PREFIX("trace.", null, PropertyType.PREFIX, "Properties in this category affect the behavior of distributed tracing."), TRACE_PORT("trace.port.client", "12234", PropertyType.PORT, "The listening port for the trace server"), TRACE_TABLE("trace.table", "trace", PropertyType.STRING, "The name of the table to store distributed traces"), @@ -454,8 +455,7 @@ public enum Property { } /** - * Gets the default value for this property exactly as provided in its - * definition (i.e., without interpolation or conversion to absolute paths). + * Gets the default value for this property exactly as provided in its definition (i.e., without interpolation or conversion to absolute paths). * * @return raw default value */ @@ -464,11 +464,9 @@ public enum Property { } /** - * Gets the default value for this property. System properties are - * interpolated into the value if necessary. + * Gets the default value for this property. System properties are interpolated into the value if necessary. * * @return default value - * @see Interpolated */ public String getDefaultValue() { if (isInterpolated()) { @@ -511,7 +509,6 @@ public enum Property { * Checks if this property is experimental. * * @return true if this property is experimental - * @see Experimental */ public boolean isExperimental() { return hasAnnotation(Experimental.class) || hasPrefixWithAnnotation(getKey(), Experimental.class); @@ -530,20 +527,18 @@ public enum Property { * Checks if this property is sensitive. * * @return true if this property is sensitive - * @see Sensitive */ public boolean isSensitive() { return hasAnnotation(Sensitive.class) || hasPrefixWithAnnotation(getKey(), Sensitive.class); } /** - * Checks if a property with the given key is sensitive. The key must be for - * a valid property, and must either itself be annotated as sensitive or have - * a prefix annotated as sensitive. + * Checks if a property with the given key is sensitive. The key must be for a valid property, and must either itself be annotated as sensitive or have a + * prefix annotated as sensitive. * - * @param key property key + * @param key + * property key * @return true if property is sensitive - * @see Sensitive */ public static boolean isSensitive(String key) { return hasPrefixWithAnnotation(key, Sensitive.class); @@ -593,11 +588,11 @@ public enum Property { } /** - * Checks if the given property key is valid. A valid property key is either - * equal to the key of some defined property or has a prefix matching some - * prefix defined in this class. + * Checks if the given property key is valid. A valid property key is either equal to the key of some defined property or has a prefix matching some prefix + * defined in this class. * - * @param key property key + * @param key + * property key * @return true if key is valid (recognized, or has a recognized prefix) */ public synchronized static boolean isValidPropertyKey(String key) { @@ -618,15 +613,13 @@ public enum Property { } /** - * Checks if the given property key is for a valid table property. A valid - * table property key is either equal to the key of some defined table - * property (which each start with {@link #TABLE_PREFIX}) or has a prefix - * matching {@link #TABLE_CONSTRAINT_PREFIX}, {@link #TABLE_ITERATOR_PREFIX}, - * or {@link #TABLE_LOCALITY_GROUP_PREFIX}. + * Checks if the given property key is for a valid table property. A valid table property key is either equal to the key of some defined table property (which + * each start with {@link #TABLE_PREFIX}) or has a prefix matching {@link #TABLE_CONSTRAINT_PREFIX}, {@link #TABLE_ITERATOR_PREFIX}, or + * {@link #TABLE_LOCALITY_GROUP_PREFIX}. * - * @param key property key - * @return true if key is valid for a table property (recognized, or has a - * recognized prefix) + * @param key + * property key + * @return true if key is valid for a table property (recognized, or has a recognized prefix) */ public synchronized static boolean isValidTablePropertyKey(String key) { if (validTableProperties == null) { @@ -647,12 +640,11 @@ public enum Property { Property.TSERV_SCAN_MAX_OPENFILES, Property.MASTER_CLIENTPORT, Property.GC_PORT); /** - * Checks if the given property may be changed via Zookeeper, but not - * recognized until the restart of some relevant daemon. + * Checks if the given property may be changed via Zookeeper, but not recognized until the restart of some relevant daemon. * - * @param key property key - * @return true if property may be changed via Zookeeper but only heeded upon - * some restart + * @param key + * property key + * @return true if property may be changed via Zookeeper but only heeded upon some restart */ public static boolean isFixedZooPropertyKey(Property key) { return fixedProperties.contains(key); @@ -669,10 +661,10 @@ public enum Property { } /** - * Checks if the given property key is valid for a property that may be - * changed via Zookeeper. + * Checks if the given property key is valid for a property that may be changed via Zookeeper. * - * @param key property key + * @param key + * property key * @return true if key's property may be changed via Zookeeper */ public static boolean isValidZooPropertyKey(String key) { @@ -686,7 +678,8 @@ public enum Property { /** * Gets a {@link Property} instance with the given key. * - * @param key property key + * @param key + * property key * @return property, or null if not found */ public static Property getPropertyByKey(String key) { @@ -713,10 +706,14 @@ public enum Property { /** * Creates a new instance of a class specified in a configuration property. * - * @param conf configuration containing property - * @param property property specifying class name - * @param base base class of type - * @param defaultInstance instance to use if creation fails + * @param conf + * configuration containing property + * @param property + * property specifying class name + * @param base + * base class of type + * @param defaultInstance + * instance to use if creation fails * @return new class instance, or default instance if creation failed * @see AccumuloVFSClassLoader */ @@ -741,14 +738,12 @@ public enum Property { } /** - * Collects together properties from the given configuration pertaining to - * compaction strategies. The relevant properties all begin with the prefix - * in {@link #TABLE_COMPACTION_STRATEGY_PREFIX}. In the returned map, the - * prefix is removed from each property's key. + * Collects together properties from the given configuration pertaining to compaction strategies. The relevant properties all begin with the prefix in + * {@link #TABLE_COMPACTION_STRATEGY_PREFIX}. In the returned map, the prefix is removed from each property's key. * - * @param tableConf configuration - * @return map of compaction strategy property keys and values, with the - * detection prefix removed from each key + * @param tableConf + * configuration + * @return map of compaction strategy property keys and values, with the detection prefix removed from each key */ public static Map<String,String> getCompactionStrategyOptions(AccumuloConfiguration tableConf) { Map<String,String> longNames = tableConf.getAllPropertiesWithPrefix(Property.TABLE_COMPACTION_STRATEGY_PREFIX); http://git-wip-us.apache.org/repos/asf/accumulo/blob/dcc19ccb/core/src/test/java/org/apache/accumulo/core/util/ValidatorTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/accumulo/core/util/ValidatorTest.java b/core/src/test/java/org/apache/accumulo/core/util/ValidatorTest.java index 6fdeabc..2be05f0 100644 --- a/core/src/test/java/org/apache/accumulo/core/util/ValidatorTest.java +++ b/core/src/test/java/org/apache/accumulo/core/util/ValidatorTest.java @@ -16,10 +16,12 @@ */ package org.apache.accumulo.core.util; -import java.util.regex.Pattern; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + import org.junit.Before; import org.junit.Test; -import static org.junit.Assert.*; public class ValidatorTest { private static class TestValidator extends Validator<String> { http://git-wip-us.apache.org/repos/asf/accumulo/blob/dcc19ccb/fate/src/main/java/org/apache/accumulo/fate/Fate.java ---------------------------------------------------------------------- diff --git a/fate/src/main/java/org/apache/accumulo/fate/Fate.java b/fate/src/main/java/org/apache/accumulo/fate/Fate.java index 7fe1142..8db7469 100644 --- a/fate/src/main/java/org/apache/accumulo/fate/Fate.java +++ b/fate/src/main/java/org/apache/accumulo/fate/Fate.java @@ -26,28 +26,28 @@ import org.apache.log4j.Logger; /** * Fault tolerant executor - * - * + * + * */ public class Fate<T> { - + private static final String DEBUG_PROP = "debug"; private static final String AUTO_CLEAN_PROP = "autoClean"; private static final String EXCEPTION_PROP = "exception"; private static final String RETURN_PROP = "return"; - + final private static Logger log = Logger.getLogger(Fate.class); - + private TStore<T> store; private T environment; - + private static final EnumSet<TStatus> FINISHED_STATES = EnumSet.of(TStatus.FAILED, TStatus.SUCCESSFUL, TStatus.UNKNOWN); - + private AtomicBoolean keepRunning = new AtomicBoolean(true); - + private class TransactionRunner implements Runnable { - + @Override public void run() { while (keepRunning.get()) { @@ -67,12 +67,12 @@ public class Fate<T> { op = op.call(tid, environment); } else continue; - + } catch (Exception e) { transitionToFailed(tid, op, e); continue; } - + if (op == null) { // transaction is finished String ret = prevOp.getReturn(); @@ -94,10 +94,10 @@ public class Fate<T> { } finally { store.unreserve(tid, deferTime); } - + } } - + private void transitionToFailed(long tid, Repo<T> op, Exception e) { String tidStr = String.format("%016x", tid); log.warn("Failed to execute Repo, tid=" + tidStr, e); @@ -105,19 +105,19 @@ public class Fate<T> { store.setStatus(tid, TStatus.FAILED_IN_PROGRESS); log.info("Updated status for Repo with tid=" + tidStr + " to FAILED_IN_PROGRESS"); } - + private void processFailed(long tid, Repo<T> op) { while (op != null) { undo(tid, op); - + store.pop(tid); op = store.top(tid); } - + store.setStatus(tid, TStatus.FAILED); doCleanUp(tid); } - + private void doCleanUp(long tid) { Boolean autoClean = (Boolean) store.getProperty(tid, AUTO_CLEAN_PROP); if (autoClean != null && autoClean) { @@ -129,7 +129,7 @@ public class Fate<T> { store.pop(tid); } } - + private void undo(long tid, Repo<T> op) { try { op.undo(tid, environment); @@ -137,14 +137,13 @@ public class Fate<T> { log.warn("Failed to undo Repo, tid=" + String.format("%016x", tid), e); } } - + } - + /** * Creates a Fault-tolerant executor. * <p> - * Note: Users of this class should call {@link #startTransactionRunners(int)} to - * launch the {@link TransactionRunner} threads after creating a Fate object. + * Note: Users of this class should call {@link #startTransactionRunners(int)} to launch the worker threads after creating a Fate object. */ public Fate(T environment, TStore<T> store) { this.store = store; @@ -152,7 +151,7 @@ public class Fate<T> { } /** - * Launches the specified number of {@link TransactionRunner} threads. + * Launches the specified number of worker threads. */ public void startTransactionRunners(int numThreads) { for (int i = 0; i < numThreads; i++) { @@ -161,13 +160,13 @@ public class Fate<T> { thread.start(); } } - + // get a transaction id back to the requester before doing any work public long startTransaction() { long dir = store.create(); return dir; } - + // start work in the transaction.. it is safe to call this // multiple times for a transaction... but it will only seed once public void seedTransaction(long tid, Repo<T> repo, boolean autoCleanUp) { @@ -182,25 +181,25 @@ public class Fate<T> { throw new RuntimeException(e); } } - + if (autoCleanUp) store.setProperty(tid, AUTO_CLEAN_PROP, Boolean.valueOf(autoCleanUp)); - + store.setProperty(tid, DEBUG_PROP, repo.getDescription()); - + store.setStatus(tid, TStatus.IN_PROGRESS); } } finally { store.unreserve(tid, 0); } - + } - + // check on the transaction public TStatus waitForCompletion(long tid) { return store.waitForStatusChange(tid, FINISHED_STATES); } - + // resource cleanup public void delete(long tid) { store.reserve(tid); @@ -222,7 +221,7 @@ public class Fate<T> { store.unreserve(tid, 0); } } - + public String getReturn(long tid) { store.reserve(tid); try { @@ -233,7 +232,7 @@ public class Fate<T> { store.unreserve(tid, 0); } } - + // get reportable failures public Exception getException(long tid) { store.reserve(tid); @@ -245,7 +244,7 @@ public class Fate<T> { store.unreserve(tid, 0); } } - + /** * Flags that FATE threadpool to clear out and end. Does not actively stop running FATE processes. */ http://git-wip-us.apache.org/repos/asf/accumulo/blob/dcc19ccb/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java ---------------------------------------------------------------------- diff --git a/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java b/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java index d0ca27f..ac3426e 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java +++ b/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java @@ -43,12 +43,6 @@ public class MonitorLog4jWatcher extends FileWatchdog implements Watcher { private boolean loggingDisabled = false; protected String path; - /** - * @param zkPath - * @param filename - * @param delay - * @param propertyName - */ public MonitorLog4jWatcher(String instance, String filename, int delay) { super(filename); setDelay(delay); @@ -78,7 +72,7 @@ public class MonitorLog4jWatcher extends FileWatchdog implements Watcher { resetLogger(); return; } - + synchronized (lock) { // We might triggered by file-reloading or from ZK update. // Either way will result in log-forwarding being restarted @@ -87,7 +81,7 @@ public class MonitorLog4jWatcher extends FileWatchdog implements Watcher { resetLogger(); } } - + private void resetLogger() { // Force a reset on the logger's configuration LogManager.resetConfiguration(); http://git-wip-us.apache.org/repos/asf/accumulo/blob/dcc19ccb/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java ---------------------------------------------------------------------- diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java index c88ee1e..3d1f90e 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java @@ -36,7 +36,6 @@ import org.apache.accumulo.core.data.Mutation; import org.apache.accumulo.core.util.UtilWaitThread; import org.apache.accumulo.server.conf.TableConfiguration; import org.apache.accumulo.server.fs.VolumeManager; -import org.apache.accumulo.tserver.Tablet; import org.apache.accumulo.tserver.Tablet.CommitSession; import org.apache.accumulo.tserver.TabletMutations; import org.apache.accumulo.tserver.TabletServer; @@ -46,23 +45,23 @@ import org.apache.log4j.Logger; /** * Central logging facility for the TServerInfo. - * + * * Forwards in-memory updates to remote logs, carefully writing the same data to every log, while maintaining the maximum thread parallelism for greater * performance. As new logs are used and minor compactions are performed, the metadata table is kept up-to-date. - * + * */ public class TabletServerLogger { - + private static final Logger log = Logger.getLogger(TabletServerLogger.class); - + private final AtomicLong logSizeEstimate = new AtomicLong(); private final long maxSize; - + private final TabletServer tserver; - + // The current log set: always updated to a new set with every change of loggers private final List<DfsLogger> loggers = new ArrayList<DfsLogger>(); - + // The current generation of logSet. // Because multiple threads can be using a log set at one time, a log // failure is likely to affect multiple threads, who will all attempt to @@ -71,29 +70,29 @@ public class TabletServerLogger { // We'll use this generational counter to determine if another thread has // already fetched a new logSet. private AtomicInteger logSetId = new AtomicInteger(); - + // Use a ReadWriteLock to allow multiple threads to use the log set, but obtain a write lock to change them private final ReentrantReadWriteLock logSetLock = new ReentrantReadWriteLock(); - + private final AtomicInteger seqGen = new AtomicInteger(); - + private static boolean enabled(TableConfiguration tconf) { return tconf.getBoolean(Property.TABLE_WALOG_ENABLED); } - + private static boolean enabled(CommitSession commitSession) { return enabled(commitSession.getTablet().getTableConfiguration()); } - + static private abstract class TestCallWithWriteLock { abstract boolean test(); - + abstract void withWriteLock() throws IOException; } - + /** * Pattern taken from the documentation for ReentrantReadWriteLock - * + * * @param rwlock * lock to use * @param code @@ -128,15 +127,16 @@ public class TabletServerLogger { rwlock.readLock().unlock(); } } - + public TabletServerLogger(TabletServer tserver, long maxSize) { this.tserver = tserver; this.maxSize = maxSize; } - + private int initializeLoggers(final List<DfsLogger> copy) throws IOException { final int[] result = {-1}; testLockAndRun(logSetLock, new TestCallWithWriteLock() { + @Override boolean test() { copy.clear(); copy.addAll(loggers); @@ -144,7 +144,8 @@ public class TabletServerLogger { result[0] = logSetId.get(); return loggers.isEmpty(); } - + + @Override void withWriteLock() throws IOException { try { createLoggers(); @@ -161,7 +162,7 @@ public class TabletServerLogger { }); return result[0]; } - + public void getLogFiles(Set<String> loggersOut) { logSetLock.readLock().lock(); try { @@ -172,16 +173,16 @@ public class TabletServerLogger { logSetLock.readLock().unlock(); } } - + synchronized private void createLoggers() throws IOException { if (!logSetLock.isWriteLockedByCurrentThread()) { throw new IllegalStateException("createLoggers should be called with write lock held!"); } - + if (loggers.size() != 0) { throw new IllegalStateException("createLoggers should not be called when loggers.size() is " + loggers.size()); } - + try { DfsLogger alog = new DfsLogger(tserver.getServerConfig()); alog.open(tserver.getClientAddressString()); @@ -192,7 +193,7 @@ public class TabletServerLogger { throw new RuntimeException(t); } } - + public void resetLoggers() throws IOException { logSetLock.writeLock().lock(); try { @@ -201,7 +202,7 @@ public class TabletServerLogger { logSetLock.writeLock().unlock(); } } - + synchronized private void close() throws IOException { if (!logSetLock.isWriteLockedByCurrentThread()) { throw new IllegalStateException("close should be called with write lock held!"); @@ -222,22 +223,22 @@ public class TabletServerLogger { throw new IOException(t); } } - + interface Writer { LoggerOperation write(DfsLogger logger, int seq) throws Exception; } - + private int write(CommitSession commitSession, boolean mincFinish, Writer writer) throws IOException { List<CommitSession> sessions = Collections.singletonList(commitSession); return write(sessions, mincFinish, writer); } - + private int write(Collection<CommitSession> sessions, boolean mincFinish, Writer writer) throws IOException { // Work very hard not to lock this during calls to the outside world int currentLogSet = logSetId.get(); - + int seq = -1; - + int attempt = 0; boolean success = false; while (!success) { @@ -245,10 +246,10 @@ public class TabletServerLogger { // get a reference to the loggers that no other thread can touch ArrayList<DfsLogger> copy = new ArrayList<DfsLogger>(); currentLogSet = initializeLoggers(copy); - + // add the logger to the log set for the memory in the tablet, // update the metadata table if we've never used this tablet - + if (currentLogSet == logSetId.get()) { for (CommitSession commitSession : sessions) { if (commitSession.beginUpdatingLogsUsed(copy, mincFinish)) { @@ -263,10 +264,10 @@ public class TabletServerLogger { } } } - + // Make sure that the logs haven't changed out from underneath our copy if (currentLogSet == logSetId.get()) { - + // write the mutation to the logs seq = seqGen.incrementAndGet(); if (seq < 0) @@ -277,11 +278,11 @@ public class TabletServerLogger { if (lop != null) queuedOperations.add(lop); } - + for (LoggerOperation lop : queuedOperations) { lop.await(); } - + // double-check: did the log set change? success = (currentLogSet == logSetId.get()); } @@ -299,12 +300,12 @@ public class TabletServerLogger { final int finalCurrent = currentLogSet; if (!success) { testLockAndRun(logSetLock, new TestCallWithWriteLock() { - + @Override boolean test() { return finalCurrent == logSetId.get(); } - + @Override void withWriteLock() throws IOException { close(); @@ -315,17 +316,19 @@ public class TabletServerLogger { // if the log gets too big, reset it .. grab the write lock first logSizeEstimate.addAndGet(4 * 3); // event, tid, seq overhead testLockAndRun(logSetLock, new TestCallWithWriteLock() { + @Override boolean test() { return logSizeEstimate.get() > maxSize; } - + + @Override void withWriteLock() throws IOException { close(); } }); return seq; } - + public int defineTablet(final CommitSession commitSession) throws IOException { // scribble this into the metadata tablet, too. if (!enabled(commitSession)) @@ -338,7 +341,7 @@ public class TabletServerLogger { } }); } - + public int log(final CommitSession commitSession, final int tabletSeq, final Mutation m) throws IOException { if (!enabled(commitSession)) return -1; @@ -351,9 +354,9 @@ public class TabletServerLogger { logSizeEstimate.addAndGet(m.numBytes()); return seq; } - + public int logManyTablets(Map<CommitSession,List<Mutation>> mutations) throws IOException { - + final Map<CommitSession,List<Mutation>> loggables = new HashMap<CommitSession,List<Mutation>>(mutations); for (CommitSession t : mutations.keySet()) { if (!enabled(t)) @@ -361,7 +364,7 @@ public class TabletServerLogger { } if (loggables.size() == 0) return -1; - + int seq = write(loggables.keySet(), false, new Writer() { @Override public LoggerOperation write(DfsLogger logger, int ignored) throws Exception { @@ -382,14 +385,14 @@ public class TabletServerLogger { } return seq; } - + public void minorCompactionFinished(final CommitSession commitSession, final String fullyQualifiedFileName, final int walogSeq) throws IOException { - + if (!enabled(commitSession)) return; - + long t1 = System.currentTimeMillis(); - + int seq = write(commitSession, true, new Writer() { @Override public LoggerOperation write(DfsLogger logger, int ignored) throws Exception { @@ -397,12 +400,12 @@ public class TabletServerLogger { return null; } }); - + long t2 = System.currentTimeMillis(); - + log.debug(" wrote MinC finish " + seq + ": writeTime:" + (t2 - t1) + "ms "); } - + public int minorCompactionStarted(final CommitSession commitSession, final int seq, final String fullyQualifiedFileName) throws IOException { if (!enabled(commitSession)) return -1; @@ -415,8 +418,9 @@ public class TabletServerLogger { }); return seq; } - - public void recover(VolumeManager fs, KeyExtent extent, TableConfiguration tconf, List<Path> logs, Set<String> tabletFiles, MutationReceiver mr) throws IOException { + + public void recover(VolumeManager fs, KeyExtent extent, TableConfiguration tconf, List<Path> logs, Set<String> tabletFiles, MutationReceiver mr) + throws IOException { if (!enabled(tconf)) return; try { @@ -426,5 +430,5 @@ public class TabletServerLogger { throw new IOException(e); } } - + } http://git-wip-us.apache.org/repos/asf/accumulo/blob/dcc19ccb/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletResourceManagerTest.java ---------------------------------------------------------------------- diff --git a/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletResourceManagerTest.java b/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletResourceManagerTest.java index 9d9db6f..baa6f7b 100644 --- a/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletResourceManagerTest.java +++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletResourceManagerTest.java @@ -16,26 +16,24 @@ */ package org.apache.accumulo.tserver; +import static org.easymock.EasyMock.createMock; +import static org.junit.Assert.assertEquals; + import org.apache.accumulo.core.data.KeyExtent; import org.apache.accumulo.server.conf.TableConfiguration; import org.apache.accumulo.tserver.TabletServerResourceManager.TabletResourceManager; import org.junit.Before; import org.junit.Test; -import static org.junit.Assert.*; -import org.easymock.Capture; -import static org.easymock.EasyMock.*; public class TabletResourceManagerTest { private TabletServerResourceManager tsrm; private TabletResourceManager trm; - private Tablet tablet; private KeyExtent extent; private TableConfiguration conf; @Before public void setUp() throws Exception { tsrm = createMock(TabletServerResourceManager.class); - tablet = createMock(Tablet.class); extent = createMock(KeyExtent.class); conf = createMock(TableConfiguration.class); trm = tsrm.new TabletResourceManager(extent, conf);
