accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ctubb...@apache.org
Subject [04/50] [abbrv] Merge branch '1.5' into 1.6
Date Sat, 01 Nov 2014 04:56:58 GMT
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
index 0e02f05,0000000..7158ea8
mode 100644,000000..100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
@@@ -1,598 -1,0 +1,598 @@@
 +/*
 + * Licensed to the Apache Software Foundation (ASF) under one or more
 + * contributor license agreements.  See the NOTICE file distributed with
 + * this work for additional information regarding copyright ownership.
 + * The ASF licenses this file to You under the Apache License, Version 2.0
 + * (the "License"); you may not use this file except in compliance with
 + * the License.  You may obtain a copy of the License at
 + *
 + *     http://www.apache.org/licenses/LICENSE-2.0
 + *
 + * Unless required by applicable law or agreed to in writing, software
 + * distributed under the License is distributed on an "AS IS" BASIS,
 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 + * See the License for the specific language governing permissions and
 + * limitations under the License.
 + */
 +package org.apache.accumulo.tserver.log;
 +
++import static com.google.common.base.Charsets.UTF_8;
 +import static org.apache.accumulo.tserver.logger.LogEvents.COMPACTION_FINISH;
 +import static org.apache.accumulo.tserver.logger.LogEvents.COMPACTION_START;
 +import static org.apache.accumulo.tserver.logger.LogEvents.DEFINE_TABLET;
 +import static org.apache.accumulo.tserver.logger.LogEvents.MANY_MUTATIONS;
 +import static org.apache.accumulo.tserver.logger.LogEvents.OPEN;
 +
 +import java.io.DataInputStream;
 +import java.io.DataOutputStream;
 +import java.io.EOFException;
 +import java.io.IOException;
 +import java.io.OutputStream;
 +import java.lang.reflect.InvocationTargetException;
 +import java.lang.reflect.Method;
 +import java.nio.channels.ClosedChannelException;
 +import java.util.ArrayList;
 +import java.util.Arrays;
 +import java.util.Collections;
 +import java.util.HashMap;
 +import java.util.List;
 +import java.util.Map;
 +import java.util.Set;
 +import java.util.UUID;
 +import java.util.concurrent.CountDownLatch;
 +import java.util.concurrent.LinkedBlockingQueue;
 +
- import org.apache.accumulo.core.Constants;
 +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;
 +import org.apache.accumulo.core.security.crypto.CryptoModule;
 +import org.apache.accumulo.core.security.crypto.CryptoModuleFactory;
 +import org.apache.accumulo.core.security.crypto.CryptoModuleParameters;
 +import org.apache.accumulo.core.security.crypto.DefaultCryptoModule;
 +import org.apache.accumulo.core.security.crypto.NoFlushOutputStream;
 +import org.apache.accumulo.core.util.Daemon;
 +import org.apache.accumulo.core.util.LoggingRunnable;
 +import org.apache.accumulo.core.util.Pair;
 +import org.apache.accumulo.core.util.StringUtil;
 +import org.apache.accumulo.server.ServerConstants;
 +import org.apache.accumulo.server.fs.VolumeManager;
 +import org.apache.accumulo.server.master.state.TServerInstance;
 +import org.apache.accumulo.tserver.TabletMutations;
 +import org.apache.accumulo.tserver.logger.LogFileKey;
 +import org.apache.accumulo.tserver.logger.LogFileValue;
 +import org.apache.hadoop.fs.FSDataInputStream;
 +import org.apache.hadoop.fs.FSDataOutputStream;
 +import org.apache.hadoop.fs.Path;
 +import org.apache.log4j.Logger;
 +
 +/**
 + * Wrap a connection to a logger.
 + *
 + */
 +public class DfsLogger {
 +  public static final String LOG_FILE_HEADER_V2 = "--- Log File Header (v2) ---";
 +  public static final String LOG_FILE_HEADER_V3 = "--- Log File Header (v3) ---";
 +
 +  private static Logger log = Logger.getLogger(DfsLogger.class);
 +
 +  public static class LogClosedException extends IOException {
 +    private static final long serialVersionUID = 1L;
 +
 +    public LogClosedException() {
 +      super("LogClosed");
 +    }
 +  }
 +
 +  /**
 +   * A well-timed tabletserver failure could result in an incomplete header written to a write-ahead log. This exception is thrown when the header cannot be
 +   * read from a WAL which should only happen when the tserver dies as described.
 +   */
 +  public static class LogHeaderIncompleteException extends IOException {
 +    private static final long serialVersionUID = 1l;
 +
 +    public LogHeaderIncompleteException(String msg) {
 +      super(msg);
 +    }
 +
 +    public LogHeaderIncompleteException(String msg, Throwable cause) {
 +      super(msg, cause);
 +    }
 +
 +    public LogHeaderIncompleteException(Throwable cause) {
 +      super(cause);
 +    }
 +  }
 +
 +  public static class DFSLoggerInputStreams {
 +
 +    private FSDataInputStream originalInput;
 +    private DataInputStream decryptingInputStream;
 +
 +    public DFSLoggerInputStreams(FSDataInputStream originalInput, DataInputStream decryptingInputStream) {
 +      this.originalInput = originalInput;
 +      this.decryptingInputStream = decryptingInputStream;
 +    }
 +
 +    public FSDataInputStream getOriginalInput() {
 +      return originalInput;
 +    }
 +
 +    public void setOriginalInput(FSDataInputStream originalInput) {
 +      this.originalInput = originalInput;
 +    }
 +
 +    public DataInputStream getDecryptingInputStream() {
 +      return decryptingInputStream;
 +    }
 +
 +    public void setDecryptingInputStream(DataInputStream decryptingInputStream) {
 +      this.decryptingInputStream = decryptingInputStream;
 +    }
 +  }
 +
 +  public interface ServerResources {
 +    AccumuloConfiguration getConfiguration();
 +
 +    VolumeManager getFileSystem();
 +
 +    Set<TServerInstance> getCurrentTServers();
 +  }
 +
 +  private final LinkedBlockingQueue<DfsLogger.LogWork> workQueue = new LinkedBlockingQueue<DfsLogger.LogWork>();
 +
 +  private final Object closeLock = new Object();
 +
 +  private static final DfsLogger.LogWork CLOSED_MARKER = new DfsLogger.LogWork(null);
 +
 +  private static final LogFileValue EMPTY = new LogFileValue();
 +
 +  private boolean closed = false;
 +
 +  private class LogSyncingTask implements Runnable {
 +
 +    @Override
 +    public void run() {
 +      ArrayList<DfsLogger.LogWork> work = new ArrayList<DfsLogger.LogWork>();
 +      boolean sawClosedMarker = false;
 +      while (!sawClosedMarker) {
 +        work.clear();
 +
 +        try {
 +          work.add(workQueue.take());
 +        } catch (InterruptedException ex) {
 +          continue;
 +        }
 +        workQueue.drainTo(work);
 +
 +        try {
 +          sync.invoke(logFile);
 +        } catch (Exception ex) {
 +          log.warn("Exception syncing " + ex);
 +          for (DfsLogger.LogWork logWork : work) {
 +            logWork.exception = ex;
 +          }
 +        }
 +
 +        for (DfsLogger.LogWork logWork : work)
 +          if (logWork == CLOSED_MARKER)
 +            sawClosedMarker = true;
 +          else
 +            logWork.latch.countDown();
 +      }
 +    }
 +  }
 +
 +  static class LogWork {
 +    CountDownLatch latch;
 +    volatile Exception exception;
 +
 +    public LogWork(CountDownLatch latch) {
 +      this.latch = latch;
 +    }
 +  }
 +
 +  public static class LoggerOperation {
 +    private final LogWork work;
 +
 +    public LoggerOperation(LogWork work) {
 +      this.work = work;
 +    }
 +
 +    public void await() throws IOException {
 +      try {
 +        work.latch.await();
 +      } catch (InterruptedException e) {
 +        throw new RuntimeException(e);
 +      }
 +
 +      if (work.exception != null) {
 +        if (work.exception instanceof IOException)
 +          throw (IOException) work.exception;
 +        else if (work.exception instanceof RuntimeException)
 +          throw (RuntimeException) work.exception;
 +        else
 +          throw new RuntimeException(work.exception);
 +      }
 +    }
 +  }
 +
 +  @Override
 +  public boolean equals(Object obj) {
 +    // filename is unique
 +    if (obj == null)
 +      return false;
 +    if (obj instanceof DfsLogger)
 +      return getFileName().equals(((DfsLogger) obj).getFileName());
 +    return false;
 +  }
 +
 +  @Override
 +  public int hashCode() {
 +    // filename is unique
 +    return getFileName().hashCode();
 +  }
 +
 +  private final ServerResources conf;
 +  private FSDataOutputStream logFile;
 +  private DataOutputStream encryptingLogFile = null;
 +  private Method sync;
 +  private String logPath;
 +  private Daemon syncThread;
 +
 +  /* Track what's actually in +r/!0 for this logger ref */
 +  private String metaReference;
 +
 +  public DfsLogger(ServerResources conf) throws IOException {
 +    this.conf = conf;
 +  }
 +
 +  /**
 +   * Refernce a pre-existing log file.
 +   *
 +   * @param meta
 +   *          the cq for the "log" entry in +r/!0
 +   */
 +  public DfsLogger(ServerResources conf, String filename, String meta) throws IOException {
 +    this.conf = conf;
 +    this.logPath = filename;
 +    metaReference = meta;
 +  }
 +
 +  public static DFSLoggerInputStreams readHeaderAndReturnStream(VolumeManager fs, Path path, AccumuloConfiguration conf) throws IOException {
 +    FSDataInputStream input = fs.open(path);
 +    DataInputStream decryptingInput = null;
 +
 +    byte[] magic = DfsLogger.LOG_FILE_HEADER_V3.getBytes();
 +    byte[] magicBuffer = new byte[magic.length];
 +    try {
 +      input.readFully(magicBuffer);
 +      if (Arrays.equals(magicBuffer, magic)) {
 +        // additional parameters it needs from the underlying stream.
 +        String cryptoModuleClassname = input.readUTF();
 +        CryptoModule cryptoModule = CryptoModuleFactory.getCryptoModule(cryptoModuleClassname);
 +
 +        // Create the parameters and set the input stream into those parameters
 +        CryptoModuleParameters params = CryptoModuleFactory.createParamsObjectFromAccumuloConfiguration(conf);
 +        params.setEncryptedInputStream(input);
 +
 +        // Create the plaintext input stream from the encrypted one
 +        params = cryptoModule.getDecryptingInputStream(params);
 +
 +        if (params.getPlaintextInputStream() instanceof DataInputStream) {
 +          decryptingInput = (DataInputStream) params.getPlaintextInputStream();
 +        } else {
 +          decryptingInput = new DataInputStream(params.getPlaintextInputStream());
 +        }
 +      } else {
 +        input.seek(0);
 +        byte[] magicV2 = DfsLogger.LOG_FILE_HEADER_V2.getBytes();
 +        byte[] magicBufferV2 = new byte[magicV2.length];
 +        input.readFully(magicBufferV2);
 +
 +        if (Arrays.equals(magicBufferV2, magicV2)) {
 +          // Log files from 1.5 dump their options in raw to the logger files. Since we don't know the class
 +          // that needs to read those files, we can make a couple of basic assumptions. Either it's going to be
 +          // the NullCryptoModule (no crypto) or the DefaultCryptoModule.
 +
 +          // If it's null, we won't have any parameters whatsoever. First, let's attempt to read
 +          // parameters
 +          Map<String,String> opts = new HashMap<String,String>();
 +          int count = input.readInt();
 +          for (int i = 0; i < count; i++) {
 +            String key = input.readUTF();
 +            String value = input.readUTF();
 +            opts.put(key, value);
 +          }
 +
 +          if (opts.size() == 0) {
 +            // NullCryptoModule, we're done
 +            decryptingInput = input;
 +          } else {
 +
 +            // The DefaultCryptoModule will want to read the parameters from the underlying file, so we will put the file back to that spot.
 +            org.apache.accumulo.core.security.crypto.CryptoModule cryptoModule = org.apache.accumulo.core.security.crypto.CryptoModuleFactory
 +                .getCryptoModule(DefaultCryptoModule.class.getName());
 +
 +            CryptoModuleParameters params = CryptoModuleFactory.createParamsObjectFromAccumuloConfiguration(conf);
 +
 +            input.seek(0);
 +            input.readFully(magicBufferV2);
 +            params.setEncryptedInputStream(input);
 +
 +            params = cryptoModule.getDecryptingInputStream(params);
 +            if (params.getPlaintextInputStream() instanceof DataInputStream) {
 +              decryptingInput = (DataInputStream) params.getPlaintextInputStream();
 +            } else {
 +              decryptingInput = new DataInputStream(params.getPlaintextInputStream());
 +            }
 +          }
 +
 +        } else {
 +
 +          input.seek(0);
 +          decryptingInput = input;
 +        }
 +
 +      }
 +    } catch (EOFException e) {
 +      log.warn("Got EOFException trying to read WAL header information, assuming the rest of the file (" + path + ") has no data.");
 +      // A TabletServer might have died before the (complete) header was written
 +      throw new LogHeaderIncompleteException(e);
 +    }
 +
 +    return new DFSLoggerInputStreams(input, decryptingInput);
 +  }
 +
 +  public synchronized void open(String address) throws IOException {
 +    String filename = UUID.randomUUID().toString();
 +    String logger = StringUtil.join(Arrays.asList(address.split(":")), "+");
 +
 +    log.debug("DfsLogger.open() begin");
 +    VolumeManager fs = conf.getFileSystem();
 +
 +    logPath = fs.choose(ServerConstants.getBaseUris()) + Path.SEPARATOR + ServerConstants.WAL_DIR + Path.SEPARATOR + logger + Path.SEPARATOR + filename;
 +
 +    metaReference = toString();
 +    try {
 +      short replication = (short) conf.getConfiguration().getCount(Property.TSERV_WAL_REPLICATION);
 +      if (replication == 0)
 +        replication = fs.getDefaultReplication(new Path(logPath));
 +      long blockSize = conf.getConfiguration().getMemoryInBytes(Property.TSERV_WAL_BLOCKSIZE);
 +      if (blockSize == 0)
 +        blockSize = (long) (conf.getConfiguration().getMemoryInBytes(Property.TSERV_WALOG_MAX_SIZE) * 1.1);
 +      if (conf.getConfiguration().getBoolean(Property.TSERV_WAL_SYNC))
 +        logFile = fs.createSyncable(new Path(logPath), 0, replication, blockSize);
 +      else
 +        logFile = fs.create(new Path(logPath), true, 0, replication, blockSize);
 +
 +      String syncMethod = conf.getConfiguration().get(Property.TSERV_WAL_SYNC_METHOD);
 +      try {
 +        // hsync: send data to datanodes and sync the data to disk
 +        sync = logFile.getClass().getMethod(syncMethod);
 +      } catch (Exception ex) {
 +        log.warn("Could not find configured " + syncMethod + " method, trying to fall back to old Hadoop sync method", ex);
 +
 +        try {
 +          // sync: send data to datanodes
 +          sync = logFile.getClass().getMethod("sync");
 +        } catch (Exception e) {
 +          throw new RuntimeException(e);
 +        }
 +      }
 +
 +      // Initialize the crypto operations.
 +      org.apache.accumulo.core.security.crypto.CryptoModule cryptoModule = org.apache.accumulo.core.security.crypto.CryptoModuleFactory.getCryptoModule(conf
 +          .getConfiguration().get(Property.CRYPTO_MODULE_CLASS));
 +
 +      // Initialize the log file with a header and the crypto params used to set up this log file.
-       logFile.write(LOG_FILE_HEADER_V3.getBytes(Constants.UTF8));
++      logFile.write(LOG_FILE_HEADER_V3.getBytes(UTF_8));
 +
 +      CryptoModuleParameters params = CryptoModuleFactory.createParamsObjectFromAccumuloConfiguration(conf.getConfiguration());
 +
 +      NoFlushOutputStream nfos = new NoFlushOutputStream(logFile);
 +      params.setPlaintextOutputStream(nfos);
 +
 +      // In order to bootstrap the reading of this file later, we have to record the CryptoModule that was used to encipher it here,
 +      // so that that crypto module can re-read its own parameters.
 +
 +      logFile.writeUTF(conf.getConfiguration().get(Property.CRYPTO_MODULE_CLASS));
 +
 +      params = cryptoModule.getEncryptingOutputStream(params);
 +      OutputStream encipheringOutputStream = params.getEncryptedOutputStream();
 +
 +      // If the module just kicks back our original stream, then just use it, don't wrap it in
 +      // another data OutputStream.
 +      if (encipheringOutputStream == nfos) {
 +        log.debug("No enciphering, using raw output stream");
 +        encryptingLogFile = nfos;
 +      } else {
 +        log.debug("Enciphering found, wrapping in DataOutputStream");
 +        encryptingLogFile = new DataOutputStream(encipheringOutputStream);
 +      }
 +
 +      LogFileKey key = new LogFileKey();
 +      key.event = OPEN;
 +      key.tserverSession = filename;
 +      key.filename = filename;
 +      write(key, EMPTY);
 +      sync.invoke(logFile);
 +      log.debug("Got new write-ahead log: " + this);
 +    } catch (Exception ex) {
 +      if (logFile != null)
 +        logFile.close();
 +      logFile = null;
 +      encryptingLogFile = null;
 +      throw new IOException(ex);
 +    }
 +
 +    syncThread = new Daemon(new LoggingRunnable(log, new LogSyncingTask()));
 +    syncThread.setName("Accumulo WALog thread " + toString());
 +    syncThread.start();
 +  }
 +
 +  @Override
 +  public String toString() {
 +    String fileName = getFileName();
 +    if (fileName.contains(":"))
 +      return getLogger() + "/" + getFileName();
 +    return fileName;
 +  }
 +
 +  /**
 +   * get the cq needed to reference this logger's entry in +r/!0
 +   */
 +  public String getMeta() {
 +    if (null == metaReference) {
 +      throw new IllegalStateException("logger doesn't have meta reference. " + this);
 +    }
 +    return metaReference;
 +  }
 +
 +  public String getFileName() {
 +    return logPath.toString();
 +  }
 +
 +  public void close() throws IOException {
 +
 +    synchronized (closeLock) {
 +      if (closed)
 +        return;
 +      // after closed is set to true, nothing else should be added to the queue
 +      // CLOSED_MARKER should be the last thing on the queue, therefore when the
 +      // background thread sees the marker and exits there should be nothing else
 +      // to process... so nothing should be left waiting for the background
 +      // thread to do work
 +      closed = true;
 +      workQueue.add(CLOSED_MARKER);
 +    }
 +
 +    // wait for background thread to finish before closing log file
 +    if (syncThread != null) {
 +      try {
 +        syncThread.join();
 +      } catch (InterruptedException e) {
 +        throw new RuntimeException(e);
 +      }
 +    }
 +
 +    // expect workq should be empty at this point
 +    if (workQueue.size() != 0) {
 +      log.error("WAL work queue not empty after sync thread exited");
 +      throw new IllegalStateException("WAL work queue not empty after sync thread exited");
 +    }
 +
 +    if (encryptingLogFile != null)
 +      try {
 +        logFile.close();
 +      } catch (IOException ex) {
 +        log.error(ex);
 +        throw new LogClosedException();
 +      }
 +  }
 +
 +  public synchronized void defineTablet(int seq, int tid, KeyExtent tablet) throws IOException {
 +    // write this log to the METADATA table
 +    final LogFileKey key = new LogFileKey();
 +    key.event = DEFINE_TABLET;
 +    key.seq = seq;
 +    key.tid = tid;
 +    key.tablet = tablet;
 +    try {
 +      write(key, EMPTY);
 +      sync.invoke(logFile);
 +    } catch (IllegalArgumentException e) {
 +      log.error("Signature of sync method changed. Accumulo is likely incompatible with this version of Hadoop.");
 +      throw new RuntimeException(e);
 +    } catch (IllegalAccessException e) {
 +      log.error("Could not invoke sync method due to permission error.");
 +      throw new RuntimeException(e);
 +    } catch (InvocationTargetException e) {
 +      Throwable cause = e.getCause();
 +      if (cause instanceof IOException) {
 +        throw (IOException) cause;
 +      } else if (cause instanceof RuntimeException) {
 +        throw (RuntimeException) cause;
 +      } else if (cause instanceof Error) {
 +        throw (Error) cause;
 +      } else {
 +        // Cause is null, or some other checked exception that was added later.
 +        throw new RuntimeException(e);
 +      }
 +    }
 +  }
 +
 +  private synchronized void write(LogFileKey key, LogFileValue value) throws IOException {
 +    key.write(encryptingLogFile);
 +    value.write(encryptingLogFile);
 +    encryptingLogFile.flush();
 +  }
 +
 +  public LoggerOperation log(int seq, int tid, Mutation mutation) throws IOException {
 +    return logManyTablets(Collections.singletonList(new TabletMutations(tid, seq, Collections.singletonList(mutation))));
 +  }
 +
 +  private LoggerOperation logFileData(List<Pair<LogFileKey,LogFileValue>> keys) throws IOException {
 +    DfsLogger.LogWork work = new DfsLogger.LogWork(new CountDownLatch(1));
 +    synchronized (DfsLogger.this) {
 +      try {
 +        for (Pair<LogFileKey,LogFileValue> pair : keys) {
 +          write(pair.getFirst(), pair.getSecond());
 +        }
 +      } catch (ClosedChannelException ex) {
 +        throw new LogClosedException();
 +      } catch (Exception e) {
 +        log.error(e, e);
 +        work.exception = e;
 +      }
 +    }
 +
 +    synchronized (closeLock) {
 +      // use a different lock for close check so that adding to work queue does not need
 +      // to wait on walog I/O operations
 +
 +      if (closed)
 +        throw new LogClosedException();
 +      workQueue.add(work);
 +    }
 +
 +    return new LoggerOperation(work);
 +  }
 +
 +  public LoggerOperation logManyTablets(List<TabletMutations> mutations) throws IOException {
 +    List<Pair<LogFileKey,LogFileValue>> data = new ArrayList<Pair<LogFileKey,LogFileValue>>();
 +    for (TabletMutations tabletMutations : mutations) {
 +      LogFileKey key = new LogFileKey();
 +      key.event = MANY_MUTATIONS;
 +      key.seq = tabletMutations.getSeq();
 +      key.tid = tabletMutations.getTid();
 +      LogFileValue value = new LogFileValue();
 +      value.mutations = tabletMutations.getMutations();
 +      data.add(new Pair<LogFileKey,LogFileValue>(key, value));
 +    }
 +    return logFileData(data);
 +  }
 +
 +  public LoggerOperation minorCompactionFinished(int seq, int tid, String fqfn) throws IOException {
 +    LogFileKey key = new LogFileKey();
 +    key.event = COMPACTION_FINISH;
 +    key.seq = seq;
 +    key.tid = tid;
 +    return logFileData(Collections.singletonList(new Pair<LogFileKey,LogFileValue>(key, EMPTY)));
 +  }
 +
 +  public LoggerOperation minorCompactionStarted(int seq, int tid, String fqfn) throws IOException {
 +    LogFileKey key = new LogFileKey();
 +    key.event = COMPACTION_START;
 +    key.seq = seq;
 +    key.tid = tid;
 +    key.filename = fqfn;
 +    return logFileData(Collections.singletonList(new Pair<LogFileKey,LogFileValue>(key, EMPTY)));
 +  }
 +
 +  public String getLogger() {
 +    String parts[] = logPath.split("/");
 +    return StringUtil.join(Arrays.asList(parts[parts.length - 2].split("[+]")), ":");
 +  }
 +
 +}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogFileValue.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogFileValue.java
index 997f71b,0000000..2c807fd
mode 100644,000000..100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogFileValue.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogFileValue.java
@@@ -1,96 -1,0 +1,97 @@@
 +/*
 + * Licensed to the Apache Software Foundation (ASF) under one or more
 + * contributor license agreements.  See the NOTICE file distributed with
 + * this work for additional information regarding copyright ownership.
 + * The ASF licenses this file to You under the Apache License, Version 2.0
 + * (the "License"); you may not use this file except in compliance with
 + * the License.  You may obtain a copy of the License at
 + *
 + *     http://www.apache.org/licenses/LICENSE-2.0
 + *
 + * Unless required by applicable law or agreed to in writing, software
 + * distributed under the License is distributed on an "AS IS" BASIS,
 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 + * See the License for the specific language governing permissions and
 + * limitations under the License.
 + */
 +package org.apache.accumulo.tserver.logger;
 +
++import static com.google.common.base.Charsets.UTF_8;
++
 +import java.io.DataInput;
 +import java.io.DataOutput;
 +import java.io.IOException;
 +import java.util.ArrayList;
 +import java.util.Collections;
 +import java.util.List;
 +
- import org.apache.accumulo.core.Constants;
 +import org.apache.accumulo.core.data.ColumnUpdate;
 +import org.apache.accumulo.core.data.Mutation;
 +import org.apache.accumulo.server.data.ServerMutation;
 +import org.apache.hadoop.io.Writable;
 +
 +public class LogFileValue implements Writable {
 +  
 +  private static final List<Mutation> empty = Collections.emptyList();
 +  
 +  public List<Mutation> mutations = empty;
 +  
 +  @Override
 +  public void readFields(DataInput in) throws IOException {
 +    int count = in.readInt();
 +    mutations = new ArrayList<Mutation>(count);
 +    for (int i = 0; i < count; i++) {
 +      ServerMutation mutation = new ServerMutation();
 +      mutation.readFields(in);
 +      mutations.add(mutation);
 +    }
 +  }
 +  
 +  @Override
 +  public void write(DataOutput out) throws IOException {
 +    out.writeInt(mutations.size());
 +    for (Mutation m : mutations) {
 +      m.write(out);
 +    }
 +  }
 +  
 +  public static void print(LogFileValue value) {
 +    System.out.println(value.toString());
 +  }
 +  
 +  private static String displayLabels(byte[] labels) {
-     String s = new String(labels, Constants.UTF8);
++    String s = new String(labels, UTF_8);
 +    s = s.replace("&", " & ");
 +    s = s.replace("|", " | ");
 +    return s;
 +  }
 +  
 +  public static String format(LogFileValue lfv, int maxMutations) {
 +    if (lfv.mutations.size() == 0)
 +      return "";
 +    StringBuilder builder = new StringBuilder();
 +    builder.append(lfv.mutations.size() + " mutations:\n");
 +    int i = 0;
 +    for (Mutation m : lfv.mutations) {
 +      if (i++ >= maxMutations) {
 +        builder.append("...");
 +        break;
 +      }
-       builder.append("  ").append(new String(m.getRow(), Constants.UTF8)).append("\n");
++      builder.append("  ").append(new String(m.getRow(), UTF_8)).append("\n");
 +      for (ColumnUpdate update : m.getUpdates()) {
 +        String value = new String(update.getValue());
-         builder.append("      ").append(new String(update.getColumnFamily(), Constants.UTF8)).append(":")
-                 .append(new String(update.getColumnQualifier(), Constants.UTF8)).append(" ").append(update.hasTimestamp() ? "[user]:" : "[system]:")
++        builder.append("      ").append(new String(update.getColumnFamily(), UTF_8)).append(":")
++                .append(new String(update.getColumnQualifier(), UTF_8)).append(" ").append(update.hasTimestamp() ? "[user]:" : "[system]:")
 +                .append(update.getTimestamp()).append(" [").append(displayLabels(update.getColumnVisibility())).append("] ")
 +                .append(update.isDeleted() ? "<deleted>" : value).append("\n");
 +      }
 +    }
 +    return builder.toString();
 +  }
 +  
 +  @Override
 +  public String toString() {
 +    return format(this, 5);
 +  }
 +  
 +}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
index c3f4fd0,0000000..29eefc8
mode 100644,000000..100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
@@@ -1,178 -1,0 +1,179 @@@
 +/*
 + * Licensed to the Apache Software Foundation (ASF) under one or more
 + * contributor license agreements.  See the NOTICE file distributed with
 + * this work for additional information regarding copyright ownership.
 + * The ASF licenses this file to You under the Apache License, Version 2.0
 + * (the "License"); you may not use this file except in compliance with
 + * the License.  You may obtain a copy of the License at
 + *
 + *     http://www.apache.org/licenses/LICENSE-2.0
 + *
 + * Unless required by applicable law or agreed to in writing, software
 + * distributed under the License is distributed on an "AS IS" BASIS,
 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 + * See the License for the specific language governing permissions and
 + * limitations under the License.
 + */
 +package org.apache.accumulo.tserver.logger;
 +
++import static com.google.common.base.Charsets.UTF_8;
++
 +import java.io.DataInputStream;
 +import java.io.EOFException;
 +import java.io.IOException;
 +import java.util.ArrayList;
 +import java.util.HashSet;
 +import java.util.List;
 +import java.util.Set;
 +import java.util.regex.Matcher;
 +import java.util.regex.Pattern;
 +
- import org.apache.accumulo.core.Constants;
 +import org.apache.accumulo.core.cli.Help;
 +import org.apache.accumulo.core.data.KeyExtent;
 +import org.apache.accumulo.core.data.Mutation;
 +import org.apache.accumulo.server.conf.ServerConfiguration;
 +import org.apache.accumulo.server.fs.VolumeManager;
 +import org.apache.accumulo.server.fs.VolumeManagerImpl;
 +import org.apache.accumulo.tserver.log.DfsLogger;
 +import org.apache.accumulo.tserver.log.DfsLogger.DFSLoggerInputStreams;
 +import org.apache.accumulo.tserver.log.DfsLogger.LogHeaderIncompleteException;
 +import org.apache.accumulo.tserver.log.MultiReader;
 +import org.apache.hadoop.fs.Path;
 +import org.apache.hadoop.io.Text;
 +import org.apache.log4j.Logger;
 +
 +import com.beust.jcommander.JCommander;
 +import com.beust.jcommander.Parameter;
 +
 +public class LogReader {
 +  private static final Logger log = Logger.getLogger(LogReader.class);
 +
 +  static class Opts extends Help {
 +    @Parameter(names = "-r", description = "print only mutations associated with the given row")
 +    String row;
 +    @Parameter(names = "-m", description = "limit the number of mutations printed per row")
 +    int maxMutations = 5;
 +    @Parameter(names = "-t", description = "print only mutations that fall within the given key extent")
 +    String extent;
 +    @Parameter(names = "-p", description = "search for a row that matches the given regex")
 +    String regexp;
 +    @Parameter(description = "<logfile> { <logfile> ... }")
 +    List<String> files = new ArrayList<String>();
 +  }
 +
 +  /**
 +   * Dump a Log File (Map or Sequence) to stdout. Will read from HDFS or local file system.
 +   *
 +   * @param args
 +   *          - first argument is the file to print
 +   */
 +  public static void main(String[] args) throws IOException {
 +    Opts opts = new Opts();
 +    opts.parseArgs(LogReader.class.getName(), args);
 +    VolumeManager fs = VolumeManagerImpl.get();
 +
 +    Matcher rowMatcher = null;
 +    KeyExtent ke = null;
 +    Text row = null;
 +    if (opts.files.isEmpty()) {
 +      new JCommander(opts).usage();
 +      return;
 +    }
 +    if (opts.row != null)
 +      row = new Text(opts.row);
 +    if (opts.extent != null) {
 +      String sa[] = opts.extent.split(";");
 +      ke = new KeyExtent(new Text(sa[0]), new Text(sa[1]), new Text(sa[2]));
 +    }
 +    if (opts.regexp != null) {
 +      Pattern pattern = Pattern.compile(opts.regexp);
 +      rowMatcher = pattern.matcher("");
 +    }
 +
 +    Set<Integer> tabletIds = new HashSet<Integer>();
 +
 +    for (String file : opts.files) {
 +
 +      Path path = new Path(file);
 +      LogFileKey key = new LogFileKey();
 +      LogFileValue value = new LogFileValue();
 +
 +      if (fs.isFile(path)) {
 +        // read log entries from a simple hdfs file
 +        DFSLoggerInputStreams streams;
 +        try {
 +          streams = DfsLogger.readHeaderAndReturnStream(fs, path, ServerConfiguration.getSiteConfiguration());
 +        } catch (LogHeaderIncompleteException e) {
 +          log.warn("Could not read header for " + path + ". Ignoring...");
 +          continue;
 +        }
 +        DataInputStream input = streams.getDecryptingInputStream();
 +
 +        try {
 +          while (true) {
 +            try {
 +              key.readFields(input);
 +              value.readFields(input);
 +            } catch (EOFException ex) {
 +              break;
 +            }
 +            printLogEvent(key, value, row, rowMatcher, ke, tabletIds, opts.maxMutations);
 +          }
 +        } finally {
 +          input.close();
 +        }
 +      } else {
 +        // read the log entries sorted in a map file
 +        MultiReader input = new MultiReader(fs, path);
 +        while (input.next(key, value)) {
 +          printLogEvent(key, value, row, rowMatcher, ke, tabletIds, opts.maxMutations);
 +        }
 +      }
 +    }
 +  }
 +
 +  public static void printLogEvent(LogFileKey key, LogFileValue value, Text row, Matcher rowMatcher, KeyExtent ke, Set<Integer> tabletIds, int maxMutations) {
 +
 +    if (ke != null) {
 +      if (key.event == LogEvents.DEFINE_TABLET) {
 +        if (key.tablet.equals(ke)) {
 +          tabletIds.add(key.tid);
 +        } else {
 +          return;
 +        }
 +      } else if (!tabletIds.contains(key.tid)) {
 +        return;
 +      }
 +    }
 +
 +    if (row != null || rowMatcher != null) {
 +      if (key.event == LogEvents.MUTATION || key.event == LogEvents.MANY_MUTATIONS) {
 +        boolean found = false;
 +        for (Mutation m : value.mutations) {
 +          if (row != null && new Text(m.getRow()).equals(row)) {
 +            found = true;
 +            break;
 +          }
 +
 +          if (rowMatcher != null) {
-             rowMatcher.reset(new String(m.getRow(), Constants.UTF8));
++            rowMatcher.reset(new String(m.getRow(), UTF_8));
 +            if (rowMatcher.matches()) {
 +              found = true;
 +              break;
 +            }
 +          }
 +        }
 +
 +        if (!found)
 +          return;
 +      } else {
 +        return;
 +      }
 +
 +    }
 +
 +    System.out.println(key);
 +    System.out.println(LogFileValue.format(value, maxMutations));
 +  }
 +
 +}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/NativeMapConcurrencyTest.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/NativeMapPerformanceTest.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/NativeMapStressTest.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/TestBinaryRows.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/TestBinaryRows.java
index a679c48,ef6311b..3c20858
--- a/test/src/main/java/org/apache/accumulo/test/TestBinaryRows.java
+++ b/test/src/main/java/org/apache/accumulo/test/TestBinaryRows.java
@@@ -71,98 -72,71 +72,98 @@@ public class TestBinaryRows 
      return l;
    }
    
 -  static class Opts extends ClientOnRequiredTable {
 +  public static class Opts extends ClientOnRequiredTable {
      @Parameter(names="--mode", description="either 'ingest', 'delete', 'randomLookups', 'split', 'verify', 'verifyDeleted'", required=true)
 -    String mode;
 +    public String mode;
      @Parameter(names="--start", description="the lowest numbered row")
 -    long start = 0;
 +    public long start = 0;
      @Parameter(names="--count", description="number of rows to ingest", required=true)
 -    long num = 0;
 +    public long num = 0;
    }
    
 -  public static void main(String[] args) {
 -    Opts opts = new Opts();
 -    BatchWriterOpts bwOpts = new BatchWriterOpts();
 -    ScannerOpts scanOpts = new ScannerOpts();
 -    opts.parseArgs(TestBinaryRows.class.getName(), args, scanOpts, bwOpts);
 +  public static void runTest(Connector connector, Opts opts, BatchWriterOpts bwOpts, ScannerOpts scanOpts) throws Exception {
      
 -    try {
 -      Connector connector = opts.getConnector();
 -      
 -      final Text CF = new Text("cf"), CQ = new Text("cq");
 -      final byte[] CF_BYTES = "cf".getBytes(UTF_8), CQ_BYTES = "cq".getBytes(UTF_8);
 +    final Text CF = new Text("cf"), CQ = new Text("cq");
-     final byte[] CF_BYTES = "cf".getBytes(Constants.UTF8), CQ_BYTES = "cq".getBytes(Constants.UTF8);
++    final byte[] CF_BYTES = "cf".getBytes(UTF_8), CQ_BYTES = "cq".getBytes(UTF_8);
 +    if (opts.mode.equals("ingest") || opts.mode.equals("delete")) {
 +      BatchWriter bw = connector.createBatchWriter(opts.tableName, bwOpts.getBatchWriterConfig());
 +      boolean delete = opts.mode.equals("delete");
        
 -      if (opts.mode.equals("ingest") || opts.mode.equals("delete")) {
 -        BatchWriter bw = connector.createBatchWriter(opts.tableName, bwOpts.getBatchWriterConfig());
 -        boolean delete = opts.mode.equals("delete");
 -        
 -        for (long i = 0; i < opts.num; i++) {
 -          byte[] row = encodeLong(i + opts.start);
 -          String value = "" + (i + opts.start);
 -          
 -          Mutation m = new Mutation(new Text(row));
 -          if (delete) {
 -            m.putDelete(CF, CQ);
 -          } else {
 -            m.put(CF, CQ, new Value(value.getBytes(UTF_8)));
 -          }
 -          bw.addMutation(m);
 +      for (long i = 0; i < opts.num; i++) {
 +        byte[] row = encodeLong(i + opts.start);
 +        String value = "" + (i + opts.start);
 +        
 +        Mutation m = new Mutation(new Text(row));
 +        if (delete) {
 +          m.putDelete(CF, CQ);
 +        } else {
-           m.put(CF, CQ, new Value(value.getBytes(Constants.UTF8)));
++          m.put(CF, CQ, new Value(value.getBytes(UTF_8)));
          }
 -        
 -        bw.close();
 -      } else if (opts.mode.equals("verifyDeleted")) {
 -        Scanner s = connector.createScanner(opts.tableName, opts.auths);
 -        s.setBatchSize(scanOpts.scanBatchSize);
 -        Key startKey = new Key(encodeLong(opts.start), CF_BYTES, CQ_BYTES, new byte[0], Long.MAX_VALUE);
 -        Key stopKey = new Key(encodeLong(opts.start + opts.num - 1), CF_BYTES, CQ_BYTES, new byte[0], 0);
 -        s.setBatchSize(50000);
 -        s.setRange(new Range(startKey, stopKey));
 -        
 -        for (Entry<Key,Value> entry : s) {
 -          System.err.println("ERROR : saw entries in range that should be deleted ( first value : " + entry.getValue().toString() + ")");
 -          System.err.println("exiting...");
 -          System.exit(1);
 -        }
 -        
 -      } else if (opts.mode.equals("verify")) {
 -        long t1 = System.currentTimeMillis();
 +        bw.addMutation(m);
 +      }
 +      
 +      bw.close();
 +    } else if (opts.mode.equals("verifyDeleted")) {
 +      Scanner s = connector.createScanner(opts.tableName, opts.auths);
 +      s.setBatchSize(scanOpts.scanBatchSize);
 +      Key startKey = new Key(encodeLong(opts.start), CF_BYTES, CQ_BYTES, new byte[0], Long.MAX_VALUE);
 +      Key stopKey = new Key(encodeLong(opts.start + opts.num - 1), CF_BYTES, CQ_BYTES, new byte[0], 0);
 +      s.setBatchSize(50000);
 +      s.setRange(new Range(startKey, stopKey));
 +      
 +      for (Entry<Key,Value> entry : s) {
 +        throw new Exception("ERROR : saw entries in range that should be deleted ( first value : " + entry.getValue().toString() + ")");
 +      }
 +      
 +    } else if (opts.mode.equals("verify")) {
 +      long t1 = System.currentTimeMillis();
 +      
 +      Scanner s = connector.createScanner(opts.tableName, opts.auths);
 +      Key startKey = new Key(encodeLong(opts.start), CF_BYTES, CQ_BYTES, new byte[0], Long.MAX_VALUE);
 +      Key stopKey = new Key(encodeLong(opts.start + opts.num - 1), CF_BYTES, CQ_BYTES, new byte[0], 0);
 +      s.setBatchSize(scanOpts.scanBatchSize);
 +      s.setRange(new Range(startKey, stopKey));
 +      
 +      long i = opts.start;
 +      
 +      for (Entry<Key,Value> e : s) {
 +        Key k = e.getKey();
 +        Value v = e.getValue();
 +
 +        checkKeyValue(i, k, v);
 +
 +        i++;
 +      }
 +      
 +      if (i != opts.start + opts.num) {
 +        throw new Exception("ERROR : did not see expected number of rows, saw " + (i - opts.start) + " expected " + opts.num);
 +      }
 +      
 +      long t2 = System.currentTimeMillis();
 +      
 +      System.out.printf("time : %9.2f secs%n", ((t2 - t1) / 1000.0));
 +      System.out.printf("rate : %9.2f entries/sec%n", opts.num / ((t2 - t1) / 1000.0));
 +      
 +    } else if (opts.mode.equals("randomLookups")) {
 +      int numLookups = 1000;
 +      
 +      Random r = new Random();
 +      
 +      long t1 = System.currentTimeMillis();
 +      
 +      for (int i = 0; i < numLookups; i++) {
 +        long row = ((r.nextLong() & 0x7fffffffffffffffl) % opts.num) + opts.start;
          
          Scanner s = connector.createScanner(opts.tableName, opts.auths);
 -        Key startKey = new Key(encodeLong(opts.start), CF_BYTES, CQ_BYTES, new byte[0], Long.MAX_VALUE);
 -        Key stopKey = new Key(encodeLong(opts.start + opts.num - 1), CF_BYTES, CQ_BYTES, new byte[0], 0);
          s.setBatchSize(scanOpts.scanBatchSize);
 +        Key startKey = new Key(encodeLong(row), CF_BYTES, CQ_BYTES, new byte[0], Long.MAX_VALUE);
 +        Key stopKey = new Key(encodeLong(row), CF_BYTES, CQ_BYTES, new byte[0], 0);
          s.setRange(new Range(startKey, stopKey));
          
 -        long i = opts.start;
 +        Iterator<Entry<Key,Value>> si = s.iterator();
          
 -        for (Entry<Key,Value> e : s) {
 +        if (si.hasNext()) {
 +          Entry<Key,Value> e = si.next();
            Key k = e.getKey();
            Value v = e.getValue();
            

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/TestIngest.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/TestIngest.java
index bc3ff05,5d8e268..c3e77af
--- a/test/src/main/java/org/apache/accumulo/test/TestIngest.java
+++ b/test/src/main/java/org/apache/accumulo/test/TestIngest.java
@@@ -16,16 -16,14 +16,17 @@@
   */
  package org.apache.accumulo.test;
  
+ import static com.google.common.base.Charsets.UTF_8;
+ 
 +import java.io.IOException;
  import java.util.Map.Entry;
  import java.util.Random;
  import java.util.Set;
  import java.util.TreeSet;
  
- import org.apache.accumulo.core.Constants;
  import org.apache.accumulo.core.cli.BatchWriterOpts;
 +import org.apache.accumulo.core.client.AccumuloException;
 +import org.apache.accumulo.core.client.AccumuloSecurityException;
  import org.apache.accumulo.core.client.BatchWriter;
  import org.apache.accumulo.core.client.Connector;
  import org.apache.accumulo.core.client.Instance;

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/TestMultiTableIngest.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/TestMultiTableIngest.java
index 7a848a3,125ef5f..16f0b3f
--- a/test/src/main/java/org/apache/accumulo/test/TestMultiTableIngest.java
+++ b/test/src/main/java/org/apache/accumulo/test/TestMultiTableIngest.java
@@@ -16,11 -16,11 +16,12 @@@
   */
  package org.apache.accumulo.test;
  
+ import static com.google.common.base.Charsets.UTF_8;
+ 
  import java.util.ArrayList;
 +import java.util.List;
  import java.util.Map.Entry;
  
- import org.apache.accumulo.core.Constants;
  import org.apache.accumulo.core.cli.BatchWriterOpts;
  import org.apache.accumulo.core.cli.ScannerOpts;
  import org.apache.accumulo.core.client.AccumuloException;
@@@ -101,8 -97,8 +102,8 @@@ public class TestMultiTableIngest 
        
        // populate
        for (int i = 0; i < opts.count; i++) {
 -        Mutation m = new Mutation(new Text(String.format("%05d", i)));
 +        Mutation m = new Mutation(new Text(String.format("%06d", i)));
-         m.put(new Text("col" + Integer.toString((i % 3) + 1)), new Text("qual"), new Value("junk".getBytes(Constants.UTF8)));
+         m.put(new Text("col" + Integer.toString((i % 3) + 1)), new Text("qual"), new Value("junk".getBytes(UTF_8)));
          b.getBatchWriter(tableNames.get(i % tableNames.size())).addMutation(m);
        }
        try {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousMoru.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousVerify.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
index 6082ebe,5962220..4f88c1b
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
@@@ -26,7 -28,7 +28,6 @@@ import java.util.Map.Entry
  import java.util.Random;
  import java.util.zip.CRC32;
  
--import org.apache.accumulo.core.Constants;
  import org.apache.accumulo.core.client.Connector;
  import org.apache.accumulo.core.client.Scanner;
  import org.apache.accumulo.core.data.Key;

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/continuous/UndefinedAnalyzer.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/UndefinedAnalyzer.java
index 4bd07ca,6124a3d..11d19c5
--- a/test/src/main/java/org/apache/accumulo/test/continuous/UndefinedAnalyzer.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/UndefinedAnalyzer.java
@@@ -82,9 -82,9 +83,9 @@@ public class UndefinedAnalyzer 
          parseLog(log);
        }
      }
 -    
 +
      private void parseLog(File log) throws Exception {
-       BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(log), Constants.UTF8));
+       BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(log), UTF_8));
        String line;
        TreeMap<Long,Long> tm = null;
        try {
@@@ -173,10 -172,10 +174,10 @@@
        SimpleDateFormat sdf = new SimpleDateFormat("dd HH:mm:ss,SSS yyyy MM");
        String currentYear = (Calendar.getInstance().get(Calendar.YEAR)) + "";
        String currentMonth = (Calendar.getInstance().get(Calendar.MONTH) + 1) + "";
 -      
 +
        for (File masterLog : masterLogs) {
 -        
 +
-         BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(masterLog), Constants.UTF8));
+         BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(masterLog), UTF_8));
          String line;
          try {
            while ((line = reader.readLine()) != null) {
@@@ -256,11 -252,11 +257,11 @@@
    public static void main(String[] args) throws Exception {
      Opts opts = new Opts();
      BatchScannerOpts bsOpts = new BatchScannerOpts();
 -    opts.parseArgs(UndefinedAnalyzer.class.getName(), args, opts);
 -    
 +    opts.parseArgs(UndefinedAnalyzer.class.getName(), args, bsOpts);
 +
      List<UndefinedNode> undefs = new ArrayList<UndefinedNode>();
 -    
 +
-     BufferedReader reader = new BufferedReader(new InputStreamReader(System.in, Constants.UTF8));
+     BufferedReader reader = new BufferedReader(new InputStreamReader(System.in, UTF_8));
      String line;
      while ((line = reader.readLine()) != null) {
        String[] tokens = line.split("\\s");

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
index f26c8d7,166e2ad..949fc52
--- a/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java
@@@ -16,6 -16,10 +16,8 @@@
   */
  package org.apache.accumulo.test.functional;
  
+ import static com.google.common.base.Charsets.UTF_8;
+ 
 -import java.net.InetAddress;
 -import java.net.InetSocketAddress;
  import java.util.HashMap;
  import java.util.Random;
  

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/performance/metadata/MetadataBatchScanTest.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/performance/metadata/MetadataBatchScanTest.java
index cb17340,33ef0b5..72efee3
--- a/test/src/main/java/org/apache/accumulo/test/performance/metadata/MetadataBatchScanTest.java
+++ b/test/src/main/java/org/apache/accumulo/test/performance/metadata/MetadataBatchScanTest.java
@@@ -24,7 -26,7 +26,6 @@@ import java.util.Random
  import java.util.TreeSet;
  import java.util.UUID;
  
--import org.apache.accumulo.core.Constants;
  import org.apache.accumulo.core.client.BatchScanner;
  import org.apache.accumulo.core.client.BatchWriter;
  import org.apache.accumulo.core.client.BatchWriterConfig;
@@@ -104,10 -100,10 +105,10 @@@ public class MetadataBatchScanTest 
          
          String dir = "/t-" + UUID.randomUUID();
          
-         TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(mut, new Value(dir.getBytes(Constants.UTF8)));
 -        Constants.METADATA_DIRECTORY_COLUMN.put(mut, new Value(dir.getBytes(UTF_8)));
++        TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(mut, new Value(dir.getBytes(UTF_8)));
          
          for (int i = 0; i < 5; i++) {
-           mut.put(DataFileColumnFamily.NAME, new Text(dir + "/00000_0000" + i + ".map"), new Value("10000,1000000".getBytes(Constants.UTF8)));
 -          mut.put(Constants.METADATA_DATAFILE_COLUMN_FAMILY, new Text(dir + "/00000_0000" + i + ".map"), new Value("10000,1000000".getBytes(UTF_8)));
++          mut.put(DataFileColumnFamily.NAME, new Text(dir + "/00000_0000" + i + ".map"), new Value("10000,1000000".getBytes(UTF_8)));
          }
          
          bw.addMutation(mut);

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/randomwalk/Module.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/randomwalk/Module.java
index 5756934,0191b50..e4ce3f3
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/Module.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/Module.java
@@@ -41,9 -37,7 +43,8 @@@ import javax.xml.parsers.DocumentBuilde
  import javax.xml.validation.Schema;
  import javax.xml.validation.SchemaFactory;
  
- import org.apache.accumulo.core.Constants;
  import org.apache.accumulo.core.client.security.tokens.PasswordToken;
 +import org.apache.accumulo.core.util.SimpleThreadPool;
  import org.apache.log4j.Level;
  import org.w3c.dom.Document;
  import org.w3c.dom.Element;
@@@ -201,163 -193,109 +202,163 @@@ public class Module extends Node 
      if (fixture != null) {
        fixture.setUp(state);
      }
 -    
 -    Node initNode = getNode(initNodeId);
 -    
 -    boolean test = false;
 -    if (initNode instanceof Test) {
 -      startTimer(initNode);
 -      test = true;
 -    }
 -    initNode.visit(state, getProps(initNodeId));
 -    if (test)
 -      stopTimer(initNode);
 -    
 -    state.visitedNode();
 -    // update aliases
 -    Set<String> aliases;
 -    if ((aliases = aliasMap.get(initNodeId)) != null)
 -      for (String alias : aliases) {
 -        ((Alias) nodes.get(alias)).update(initNodeId);
 -      }
 -    
 -    String curNodeId = initNodeId;
 -    int numHops = 0;
 -    long startTime = System.currentTimeMillis() / 1000;
 -    while (true) {
 -      // check if END state was reached
 -      if (curNodeId.equalsIgnoreCase("END")) {
 -        log.debug("reached END state");
 -        break;
 -      }
 -      // check if maxSec was reached
 -      long curTime = System.currentTimeMillis() / 1000;
 -      if ((curTime - startTime) > maxSec) {
 -        log.debug("reached maxSec(" + maxSec + ")");
 -        break;
 -      }
 -      // check if maxHops was reached
 -      if (numHops > maxHops) {
 -        log.debug("reached maxHops(" + maxHops + ")");
 -        break;
 -      }
 -      numHops++;
 -      
 -      if (!adjMap.containsKey(curNodeId) && !curNodeId.startsWith("alias.")) {
 -        throw new Exception("Reached node(" + curNodeId + ") without outgoing edges in module(" + this + ")");
 -      }
 -      AdjList adj = adjMap.get(curNodeId);
 -      String nextNodeId = adj.randomNeighbor();
 -      Node nextNode = getNode(nextNodeId);
 -      if (nextNode instanceof Alias) {
 -        nextNodeId = ((Alias) nextNode).getTargetId();
 -        nextNode = ((Alias) nextNode).get();
 +
 +    ExecutorService service = new SimpleThreadPool(1, "RandomWalk Runner");
 +
 +    try {
 +      Node initNode = getNode(initNodeId);
 +
 +      boolean test = false;
 +      if (initNode instanceof Test) {
 +        startTimer(initNode);
 +        test = true;
        }
 -      Properties nodeProps = getProps(nextNodeId);
 -      try {
 -        test = false;
 -        if (nextNode instanceof Test) {
 -          startTimer(nextNode);
 -          test = true;
 +      initNode.visit(state, getProps(initNodeId));
 +      if (test)
 +        stopTimer(initNode);
 +
 +      state.visitedNode();
 +      // update aliases
 +      Set<String> aliases;
 +      if ((aliases = aliasMap.get(initNodeId)) != null)
 +        for (String alias : aliases) {
 +          ((Alias) nodes.get(alias)).update(initNodeId);
          }
 -        nextNode.visit(state, nodeProps);
 -        if (test)
 -          stopTimer(nextNode);
 -      } catch (Exception e) {
 -        log.debug("Connector belongs to user: " + state.getConnector().whoami());
 -        log.debug("Exception occured at: " + System.currentTimeMillis());
 -        log.debug("Properties for node: " + nextNodeId);
 -        for (Entry<Object,Object> entry : nodeProps.entrySet()) {
 -          log.debug("  " + entry.getKey() + ": " + entry.getValue());
 +
 +      String curNodeId = initNodeId;
 +      int numHops = 0;
 +      long startTime = System.currentTimeMillis() / 1000;
 +      while (true) {
 +        // check if END state was reached
 +        if (curNodeId.equalsIgnoreCase("END")) {
 +          log.debug("reached END state");
 +          break;
          }
 -        log.debug("Overall Properties");
 -        for (Entry<Object,Object> entry : state.getProperties().entrySet()) {
 -          log.debug("  " + entry.getKey() + ": " + entry.getValue());
 +        // check if maxSec was reached
 +        long curTime = System.currentTimeMillis() / 1000;
 +        if ((curTime - startTime) > maxSec) {
 +          log.debug("reached maxSec(" + maxSec + ")");
 +          break;
          }
 -        log.debug("State information");
 -        for (String key : new TreeSet<String>(state.getMap().keySet()))  {
 -          Object value = state.getMap().get(key);
 -          String logMsg = "  " + key + ": ";
 -          if (value == null)
 -            logMsg += "null";
 -          else if (value instanceof String || value instanceof Map || value instanceof Collection || value instanceof Number)
 -            logMsg += value;
 -          else if (value instanceof byte[])
 -            logMsg += new String((byte[])value, UTF_8);
 -          else if (value instanceof PasswordToken)
 -            logMsg += new String(((PasswordToken) value).getPassword(), UTF_8);
 -          else
 -            logMsg += value.getClass()+ " - " + value;
 -          
 -          log.debug(logMsg);
 +
 +        // The number of seconds before the test should exit
 +        long secondsRemaining = maxSec - (curTime - startTime);
 +
 +        // check if maxHops was reached
 +        if (numHops > maxHops) {
 +          log.debug("reached maxHops(" + maxHops + ")");
 +          break;
          }
 -        throw new Exception("Error running node " + nextNodeId, e);
 -      }
 -      state.visitedNode();
 -      
 -      // update aliases
 -      if ((aliases = aliasMap.get(curNodeId)) != null)
 -        for (String alias : aliases) {
 -          ((Alias) nodes.get(alias)).update(curNodeId);
 +        numHops++;
 +
 +        if (!adjMap.containsKey(curNodeId) && !curNodeId.startsWith("alias.")) {
 +          throw new Exception("Reached node(" + curNodeId + ") without outgoing edges in module(" + this + ")");
 +        }
 +        AdjList adj = adjMap.get(curNodeId);
 +        String nextNodeId = adj.randomNeighbor();
 +        final Node nextNode;
 +        Node nextNodeOrAlias = getNode(nextNodeId);
 +        if (nextNodeOrAlias instanceof Alias) {
 +          nextNodeId = ((Alias) nextNodeOrAlias).getTargetId();
 +          nextNode = ((Alias) nextNodeOrAlias).get();
 +        } else {
 +          nextNode = nextNodeOrAlias;
 +        }
 +        final Properties nodeProps = getProps(nextNodeId);
 +        try {
 +          test = false;
 +          if (nextNode instanceof Test) {
 +            startTimer(nextNode);
 +            test = true;
 +          }
 +
 +          // Wrap the visit of the next node in the module in a callable that returns a thrown exception
 +          FutureTask<Exception> task = new FutureTask<Exception>(new Callable<Exception>() {
 +
 +            @Override
 +            public Exception call() throws Exception {
 +              try {
 +                nextNode.visit(state, nodeProps);
 +                return null;
 +              } catch (Exception e) {
 +                return e;
 +              }
 +            }
 +
 +          });
 +
 +          // Run the task (should execute immediately)
 +          service.submit(task);
 +
 +          Exception nodeException;
 +          try {
 +            // Bound the time we'll wait for the node to complete
 +            nodeException = task.get(secondsRemaining, TimeUnit.SECONDS);
 +          } catch (InterruptedException e) {
 +            log.warn("Interrupted waiting for " + nextNode.getClass().getSimpleName() + " to complete. Exiting.", e);
 +            break;
 +          } catch (ExecutionException e) {
 +            log.error("Caught error executing " + nextNode.getClass().getSimpleName(), e);
 +            throw e;
 +          } catch (TimeoutException e) {
 +            log.info("Timed out waiting for " + nextNode.getClass().getSimpleName() + " to complete (waited " + secondsRemaining + " seconds). Exiting.", e);
 +            break;
 +          }
 +
 +          // The RandomWalk node throw an Exception that that Callable handed back
 +          // Throw it and let the Module perform cleanup
 +          if (null != nodeException) {
 +            throw nodeException;
 +          }
 +
 +          if (test)
 +            stopTimer(nextNode);
 +        } catch (Exception e) {
 +          log.debug("Connector belongs to user: " + state.getConnector().whoami());
 +          log.debug("Exception occured at: " + System.currentTimeMillis());
 +          log.debug("Properties for node: " + nextNodeId);
 +          for (Entry<Object,Object> entry : nodeProps.entrySet()) {
 +            log.debug("  " + entry.getKey() + ": " + entry.getValue());
 +          }
 +          log.debug("Overall Properties");
 +          for (Entry<Object,Object> entry : state.getProperties().entrySet()) {
 +            log.debug("  " + entry.getKey() + ": " + entry.getValue());
 +          }
 +          log.debug("State information");
 +          for (String key : new TreeSet<String>(state.getMap().keySet())) {
 +            Object value = state.getMap().get(key);
 +            String logMsg = "  " + key + ": ";
 +            if (value == null)
 +              logMsg += "null";
 +            else if (value instanceof String || value instanceof Map || value instanceof Collection || value instanceof Number)
 +              logMsg += value;
 +            else if (value instanceof byte[])
-               logMsg += new String((byte[]) value, Constants.UTF8);
++              logMsg += new String((byte[]) value, UTF_8);
 +            else if (value instanceof PasswordToken)
-               logMsg += new String(((PasswordToken) value).getPassword(), Constants.UTF8);
++              logMsg += new String(((PasswordToken) value).getPassword(), UTF_8);
 +            else
 +              logMsg += value.getClass() + " - " + value;
 +
 +            log.debug(logMsg);
 +          }
 +          throw new Exception("Error running node " + nextNodeId, e);
          }
 -      
 -      curNodeId = nextNodeId;
 +        state.visitedNode();
 +
 +        // update aliases
 +        if ((aliases = aliasMap.get(curNodeId)) != null)
 +          for (String alias : aliases) {
 +            ((Alias) nodes.get(alias)).update(curNodeId);
 +          }
 +
 +        curNodeId = nextNodeId;
 +      }
 +    } finally {
 +      if (null != service) {
 +        service.shutdownNow();
 +      }
      }
 -    
 +
      if (teardown && (fixture != null)) {
        log.debug("tearing down module");
        fixture.tearDown(state);

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkMinusOne.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkMinusOne.java
index 1704e49,05cc4f0..67438d9
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkMinusOne.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkMinusOne.java
@@@ -20,10 -21,10 +21,10 @@@ import static com.google.common.base.Ch
  import org.apache.accumulo.core.data.Value;
  import org.apache.accumulo.test.randomwalk.State;
  
 -public class BulkMinusOne extends BulkTest {
 -  
 +public class BulkMinusOne extends BulkImportTest {
 +
-   private static final Value negOne = new Value("-1".getBytes(Constants.UTF8));
+   private static final Value negOne = new Value("-1".getBytes(UTF_8));
 -  
 +
    @Override
    protected void runLater(State state) throws Exception {
      log.info("Decrementing");

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/BatchWrite.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/BatchWrite.java
index f002274,7f9a218..3dcff6e
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/BatchWrite.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/BatchWrite.java
@@@ -51,10 -52,10 +52,10 @@@ public class BatchWrite extends Test 
        try {
          int numRows = rand.nextInt(100000);
          for (int i = 0; i < numRows; i++) {
 -          Mutation m = new Mutation(String.format("%016x", (rand.nextLong() & 0x7fffffffffffffffl)));
 -          long val = (rand.nextLong() & 0x7fffffffffffffffl);
 +          Mutation m = new Mutation(String.format("%016x", rand.nextLong() & 0x7fffffffffffffffl));
 +          long val = rand.nextLong() & 0x7fffffffffffffffl;
            for (int j = 0; j < 10; j++) {
-             m.put("cf", "cq" + j, new Value(String.format("%016x", val).getBytes(Constants.UTF8)));
+             m.put("cf", "cq" + j, new Value(String.format("%016x", val).getBytes(UTF_8)));
            }
            
            bw.addMutation(m);

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/BulkImport.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/StopTabletServer.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/StopTabletServer.java
index c8da3d8,c227a11..b3519b6
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/StopTabletServer.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/StopTabletServer.java
@@@ -47,8 -50,8 +49,8 @@@ public class StopTabletServer extends T
            Collections.sort(children);
            Stat stat = new Stat();
            byte[] data = rdr.getData(base + "/" + child + "/" + children.get(0), stat);
-           if (!"master".equals(new String(data, Constants.UTF8))) {
+           if (!"master".equals(new String(data, UTF_8))) {
 -            result.add(new TServerInstance(AddressUtil.parseAddress(child, Property.TSERV_CLIENTPORT), stat.getEphemeralOwner()));
 +            result.add(new TServerInstance(AddressUtil.parseAddress(child, false), stat.getEphemeralOwner()));
            }
          }
        } catch (KeeperException.NoNodeException ex) {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/randomwalk/image/Write.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/randomwalk/security/SecurityHelper.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/randomwalk/security/TableOp.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/randomwalk/sequential/Write.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/main/java/org/apache/accumulo/test/randomwalk/shard/BulkInsert.java
----------------------------------------------------------------------


Mime
View raw message