accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ktur...@apache.org
Subject [1/2] git commit: ACCUMULO-2049 Made shell use new instance.volumes config
Date Thu, 27 Feb 2014 00:54:01 GMT
Repository: accumulo
Updated Branches:
  refs/heads/master 160aaf893 -> 615f81e0c


ACCUMULO-2049 Made shell use new instance.volumes config


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/a8058a69
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/a8058a69
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/a8058a69

Branch: refs/heads/master
Commit: a8058a69f24ebda32e2f54e52390a98b7f7e1bf0
Parents: 96d8662
Author: Keith Turner <kturner@apache.org>
Authored: Wed Feb 26 17:58:27 2014 -0500
Committer: Keith Turner <kturner@apache.org>
Committed: Wed Feb 26 19:48:26 2014 -0500

----------------------------------------------------------------------
 .../core/client/admin/TableOperationsImpl.java  |   4 +-
 .../core/client/impl/OfflineScanner.java        |   4 +-
 .../org/apache/accumulo/core/file/FileUtil.java | 130 -------------------
 .../accumulo/core/file/VolumeConfiguration.java |  96 ++++++++++++++
 .../accumulo/core/file/rfile/PrintInfo.java     |   4 +-
 .../core/file/rfile/bcfile/PrintInfo.java       |   4 +-
 .../apache/accumulo/core/util/shell/Shell.java  |   3 +-
 .../apache/accumulo/core/zookeeper/ZooUtil.java |   4 +-
 .../core/util/shell/ShellSetInstanceTest.java   |   3 +
 .../apache/accumulo/server/ServerConstants.java |  80 ++----------
 .../accumulo/server/client/BulkImporter.java    |   5 +-
 .../accumulo/server/fs/VolumeManagerImpl.java   |   6 +-
 .../apache/accumulo/server/init/Initialize.java |  19 +--
 .../monitor/servlets/DefaultServlet.java        |   4 +-
 .../tserver/BulkFailedCopyProcessor.java        |   4 +-
 .../accumulo/test/functional/BulkFileIT.java    |   4 +-
 16 files changed, 141 insertions(+), 233 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java
b/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java
index 0245ef1..0b2f10e 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java
@@ -82,7 +82,7 @@ import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.KeyExtent;
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
-import org.apache.accumulo.core.file.FileUtil;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.iterators.IteratorUtil;
 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
@@ -1139,7 +1139,7 @@ public class TableOperationsImpl extends TableOperationsHelper {
   @SuppressWarnings("deprecation")
   private Path checkPath(String dir, String kind, String type) throws IOException, AccumuloException
{
     Path ret;
-    FileSystem fs = FileUtil.getFileSystem(dir, CachedConfiguration.getInstance(), ServerConfigurationUtil.getConfiguration(instance));
+    FileSystem fs = VolumeConfiguration.getFileSystem(dir, CachedConfiguration.getInstance(),
ServerConfigurationUtil.getConfiguration(instance));
 
     if (dir.contains(":")) {
       ret = new Path(dir);

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineScanner.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineScanner.java b/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineScanner.java
index c60e153..c90d380 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineScanner.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineScanner.java
@@ -43,7 +43,7 @@ import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.file.FileOperations;
 import org.apache.accumulo.core.file.FileSKVIterator;
-import org.apache.accumulo.core.file.FileUtil;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.IteratorUtil;
 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
@@ -306,7 +306,7 @@ class OfflineIterator implements Iterator<Entry<Key,Value>>
{
     
     // TODO need to close files - ACCUMULO-1303
     for (String file : absFiles) {
-      FileSystem fs = FileUtil.getFileSystem(file, conf, config);
+      FileSystem fs = VolumeConfiguration.getFileSystem(file, conf, config);
       FileSKVIterator reader = FileOperations.getInstance().openReader(file, false, fs, conf,
acuTableConf, null, null);
       readers.add(reader);
     }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/core/src/main/java/org/apache/accumulo/core/file/FileUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/FileUtil.java b/core/src/main/java/org/apache/accumulo/core/file/FileUtil.java
deleted file mode 100644
index 0ee16cf..0000000
--- a/core/src/main/java/org/apache/accumulo/core/file/FileUtil.java
+++ /dev/null
@@ -1,130 +0,0 @@
-/*
- * 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.core.file;
-
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.TreeMap;
-
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.KeyExtent;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.Text;
-import org.apache.log4j.Logger;
-
-public class FileUtil {
-  
-  public static class FileInfo {
-    Key firstKey = new Key();
-    Key lastKey = new Key();
-    
-    public FileInfo(Key firstKey, Key lastKey) {
-      this.firstKey = firstKey;
-      this.lastKey = lastKey;
-    }
-    
-    public Text getFirstRow() {
-      return firstKey.getRow();
-    }
-    
-    public Text getLastRow() {
-      return lastKey.getRow();
-    }
-  }
-  
-  private static final Logger log = Logger.getLogger(FileUtil.class);
-  
-  private static class MLong {
-    public MLong(long i) {
-      l = i;
-    }
-    
-    long l;
-  }
-  
-  public static Map<KeyExtent,Long> estimateSizes(AccumuloConfiguration acuConf, Path
mapFile, long fileSize, List<KeyExtent> extents, Configuration conf,
-      FileSystem fs) throws IOException {
-    
-    long totalIndexEntries = 0;
-    Map<KeyExtent,MLong> counts = new TreeMap<KeyExtent,MLong>();
-    for (KeyExtent keyExtent : extents)
-      counts.put(keyExtent, new MLong(0));
-    
-    Text row = new Text();
-    
-    FileSKVIterator index = FileOperations.getInstance().openIndex(mapFile.toString(), fs,
conf, acuConf);
-    
-    try {
-      while (index.hasTop()) {
-        Key key = index.getTopKey();
-        totalIndexEntries++;
-        key.getRow(row);
-        
-        for (Entry<KeyExtent,MLong> entry : counts.entrySet())
-          if (entry.getKey().contains(row))
-            entry.getValue().l++;
-        
-        index.next();
-      }
-    } finally {
-      try {
-        if (index != null)
-          index.close();
-      } catch (IOException e) {
-        // continue with next file
-        log.error(e, e);
-      }
-    }
-    
-    Map<KeyExtent,Long> results = new TreeMap<KeyExtent,Long>();
-    for (KeyExtent keyExtent : extents) {
-      double numEntries = counts.get(keyExtent).l;
-      if (numEntries == 0)
-        numEntries = 1;
-      long estSize = (long) ((numEntries / totalIndexEntries) * fileSize);
-      results.put(keyExtent, estSize);
-    }
-    return results;
-  }
-  
-  public static FileSystem getFileSystem(String path, Configuration conf, AccumuloConfiguration
acuconf) throws IOException {
-    if (path.contains(":"))
-      return new Path(path).getFileSystem(conf);
-    else
-      return getFileSystem(conf, acuconf);
-  }
-
-  public static FileSystem getFileSystem(Configuration conf, AccumuloConfiguration acuconf)
throws IOException {
-    String uri = acuconf.get(Property.INSTANCE_DFS_URI);
-    if ("".equals(uri))
-      return FileSystem.get(conf);
-    else
-      try {
-        return FileSystem.get(new URI(uri), conf);
-      } catch (URISyntaxException e) {
-        throw new IOException(e);
-      }
-  }
-}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/core/src/main/java/org/apache/accumulo/core/file/VolumeConfiguration.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/VolumeConfiguration.java b/core/src/main/java/org/apache/accumulo/core/file/VolumeConfiguration.java
new file mode 100644
index 0000000..af9ca1d
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/file/VolumeConfiguration.java
@@ -0,0 +1,96 @@
+package org.apache.accumulo.core.file;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.util.CachedConfiguration;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+public class VolumeConfiguration {
+
+  public static FileSystem getFileSystem(String path, Configuration conf, AccumuloConfiguration
acuconf) throws IOException {
+    if (path.contains(":"))
+      return new Path(path).getFileSystem(conf);
+    else
+      return getDefaultFilesystem(conf, acuconf);
+  }
+
+  public static FileSystem getDefaultFilesystem(Configuration conf, AccumuloConfiguration
acuconf) throws IOException {
+    String uri = acuconf.get(Property.INSTANCE_DFS_URI);
+    if ("".equals(uri))
+      return FileSystem.get(conf);
+    else
+      try {
+        return FileSystem.get(new URI(uri), conf);
+      } catch (URISyntaxException e) {
+        throw new IOException(e);
+      }
+  }
+
+  public static String getConfiguredBaseDir(AccumuloConfiguration conf) {
+    String singleNamespace = conf.get(Property.INSTANCE_DFS_DIR);
+    String dfsUri = conf.get(Property.INSTANCE_DFS_URI);
+    String baseDir;
+  
+    if (dfsUri == null || dfsUri.isEmpty()) {
+      Configuration hadoopConfig = CachedConfiguration.getInstance();
+      try {
+        baseDir = FileSystem.get(hadoopConfig).getUri().toString() + singleNamespace;
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    } else {
+      if (!dfsUri.contains(":"))
+        throw new IllegalArgumentException("Expected fully qualified URI for " + Property.INSTANCE_DFS_URI.getKey()
+ " got " + dfsUri);
+      baseDir = dfsUri + singleNamespace;
+    }
+    return baseDir;
+  }
+
+  public static String[] getConfiguredBaseDirs(AccumuloConfiguration conf) {
+    String singleNamespace = conf.get(Property.INSTANCE_DFS_DIR);
+    String ns = conf.get(Property.INSTANCE_VOLUMES);
+  
+    String configuredBaseDirs[];
+  
+    if (ns == null || ns.isEmpty()) {
+      configuredBaseDirs = new String[] {getConfiguredBaseDir(conf)};
+    } else {
+      String namespaces[] = ns.split(",");
+      String unescapedNamespaces[] = new String[namespaces.length];
+      int i = 0;
+      for (String namespace : namespaces) {
+        if (!namespace.contains(":")) {
+          throw new IllegalArgumentException("Expected fully qualified URI for " + Property.INSTANCE_VOLUMES.getKey()
+ " got " + namespace);
+        }
+  
+        try {
+          // pass through URI to unescape hex encoded chars (e.g. convert %2C to "," char)
+          unescapedNamespaces[i++] = new Path(new URI(namespace)).toString();
+        } catch (URISyntaxException e) {
+          throw new IllegalArgumentException(Property.INSTANCE_VOLUMES.getKey() + " contains
" + namespace + " which has a syntax error", e);
+        }
+      }
+  
+      configuredBaseDirs = prefix(unescapedNamespaces, singleNamespace);
+    }
+  
+    return configuredBaseDirs;
+  }
+
+  public static String[] prefix(String bases[], String suffix) {
+    if (suffix.startsWith("/"))
+      suffix = suffix.substring(1);
+    String result[] = new String[bases.length];
+    for (int i = 0; i < bases.length; i++) {
+      result[i] = bases[i] + "/" + suffix;
+    }
+    return result;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/core/src/main/java/org/apache/accumulo/core/file/rfile/PrintInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/rfile/PrintInfo.java b/core/src/main/java/org/apache/accumulo/core/file/rfile/PrintInfo.java
index 7cee0f9..4cfefad 100644
--- a/core/src/main/java/org/apache/accumulo/core/file/rfile/PrintInfo.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/PrintInfo.java
@@ -25,7 +25,7 @@ import org.apache.accumulo.core.data.ByteSequence;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
-import org.apache.accumulo.core.file.FileUtil;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.file.blockfile.impl.CachableBlockFile;
 import org.apache.accumulo.core.file.rfile.RFile.Reader;
 import org.apache.hadoop.conf.Configuration;
@@ -50,7 +50,7 @@ public class PrintInfo {
 
     @SuppressWarnings("deprecation")
     AccumuloConfiguration aconf = AccumuloConfiguration.getSiteConfiguration();
-    FileSystem hadoopFs = FileUtil.getFileSystem(conf, aconf);
+    FileSystem hadoopFs = VolumeConfiguration.getDefaultFilesystem(conf, aconf);
     FileSystem localFs  = FileSystem.getLocal(conf);
     Opts opts = new Opts();
     opts.parseArgs(PrintInfo.class.getName(), args);

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/PrintInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/PrintInfo.java
b/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/PrintInfo.java
index 4809d80..f21190e 100644
--- a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/PrintInfo.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/PrintInfo.java
@@ -22,7 +22,7 @@ import java.util.Map.Entry;
 import java.util.Set;
 
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.core.file.FileUtil;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.file.rfile.bcfile.BCFile.MetaIndexEntry;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
@@ -57,7 +57,7 @@ public class PrintInfo {
     Configuration conf = new Configuration();
     @SuppressWarnings("deprecation")
     AccumuloConfiguration siteConf = AccumuloConfiguration.getSiteConfiguration();
-    FileSystem hadoopFs = FileUtil.getFileSystem(conf, siteConf);
+    FileSystem hadoopFs = VolumeConfiguration.getDefaultFilesystem(conf, siteConf);
     FileSystem localFs = FileSystem.getLocal(conf);
     Path path = new Path(args[0]);
     FileSystem fs;

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/core/src/main/java/org/apache/accumulo/core/util/shell/Shell.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/util/shell/Shell.java b/core/src/main/java/org/apache/accumulo/core/util/shell/Shell.java
index 8ea82d4..850816c 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/shell/Shell.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/shell/Shell.java
@@ -63,6 +63,7 @@ import org.apache.accumulo.core.conf.SiteConfiguration;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.data.thrift.TConstraintViolationSummary;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.tabletserver.thrift.ConstraintViolationException;
 import org.apache.accumulo.core.trace.DistributedTrace;
 import org.apache.accumulo.core.util.BadArgumentException;
@@ -433,7 +434,7 @@ public class Shell extends ShellOptions {
     if (instanceName == null || keepers == null) {
       AccumuloConfiguration conf = SiteConfiguration.getInstance(ServerConfigurationUtil.convertClientConfig(DefaultConfiguration.getInstance(),
clientConfig));
       if (instanceName == null) {
-        Path instanceDir = new Path(conf.get(Property.INSTANCE_DFS_DIR), "instance_id");
+        Path instanceDir = new Path(VolumeConfiguration.getConfiguredBaseDirs(conf)[0], "instance_id");
         instanceId = UUID.fromString(ZooUtil.getInstanceIDFromHdfs(instanceDir, conf));
       }
       if (keepers == null) {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooUtil.java b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooUtil.java
index a062602..de1b432 100644
--- a/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooUtil.java
+++ b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooUtil.java
@@ -23,7 +23,7 @@ import java.net.UnknownHostException;
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.client.Instance;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.core.file.FileUtil;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.util.CachedConfiguration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
@@ -49,7 +49,7 @@ public class ZooUtil extends org.apache.accumulo.fate.zookeeper.ZooUtil
{
   public static String getInstanceIDFromHdfs(Path instanceDirectory, AccumuloConfiguration
conf) {
     try {
 
-      FileSystem fs = FileUtil.getFileSystem(instanceDirectory.toString(), CachedConfiguration.getInstance(),
conf);
+      FileSystem fs = VolumeConfiguration.getFileSystem(instanceDirectory.toString(), CachedConfiguration.getInstance(),
conf);
       FileStatus[] files = null;
       try {
         files = fs.listStatus(instanceDirectory);

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/core/src/test/java/org/apache/accumulo/core/util/shell/ShellSetInstanceTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/util/shell/ShellSetInstanceTest.java
b/core/src/test/java/org/apache/accumulo/core/util/shell/ShellSetInstanceTest.java
index af810ad..5a6cc8a 100644
--- a/core/src/test/java/org/apache/accumulo/core/util/shell/ShellSetInstanceTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/util/shell/ShellSetInstanceTest.java
@@ -171,7 +171,10 @@ public class ShellSetInstanceTest {
       expect(clientConf.withZkHosts("host1,host2")).andReturn(clientConf);
     }
     if (!onlyInstance) {
+      expect(clientConf.containsKey(Property.INSTANCE_VOLUMES.getKey())).andReturn(false).atLeastOnce();
       expect(clientConf.containsKey(Property.INSTANCE_DFS_DIR.getKey())).andReturn(true).atLeastOnce();
+      expect(clientConf.containsKey(Property.INSTANCE_DFS_URI.getKey())).andReturn(true).atLeastOnce();
+      expect(clientConf.getString(Property.INSTANCE_DFS_URI.getKey())).andReturn("hdfs://nn1").atLeastOnce();
       expect(clientConf.getString(Property.INSTANCE_DFS_DIR.getKey())).andReturn("/dfs").atLeastOnce();
     }
 

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java b/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
index cca869a..8983d08 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
@@ -16,7 +16,6 @@
  */
 package org.apache.accumulo.server;
 
-import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
@@ -24,16 +23,14 @@ import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.metadata.MetadataTable;
 import org.apache.accumulo.core.util.CachedConfiguration;
 import org.apache.accumulo.core.util.Pair;
 import org.apache.accumulo.core.zookeeper.ZooUtil;
 import org.apache.accumulo.server.conf.ServerConfiguration;
 import org.apache.accumulo.server.fs.VolumeUtil;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 
 public class ServerConstants {
@@ -61,65 +58,16 @@ public class ServerConstants {
 
   public static synchronized String getDefaultBaseDir() {
     if (defaultBaseDir == null) {
-      String singleNamespace = ServerConfiguration.getSiteConfiguration().get(Property.INSTANCE_DFS_DIR);
-      String dfsUri = ServerConfiguration.getSiteConfiguration().get(Property.INSTANCE_DFS_URI);
-      String baseDir;
-
-      if (dfsUri == null || dfsUri.isEmpty()) {
-        Configuration hadoopConfig = CachedConfiguration.getInstance();
-        try {
-          baseDir = FileSystem.get(hadoopConfig).getUri().toString() + singleNamespace;
-        } catch (IOException e) {
-          throw new RuntimeException(e);
-        }
-      } else {
-        if (!dfsUri.contains(":"))
-          throw new IllegalArgumentException("Expected fully qualified URI for " + Property.INSTANCE_DFS_URI.getKey()
+ " got " + dfsUri);
-        baseDir = dfsUri + singleNamespace;
-      }
-
-      defaultBaseDir = new Path(baseDir).toString();
-
+      defaultBaseDir = new Path(VolumeConfiguration.getConfiguredBaseDir(ServerConfiguration.getSiteConfiguration())).toString();
     }
 
     return defaultBaseDir;
   }
 
-  public static String[] getConfiguredBaseDirs(AccumuloConfiguration conf) {
-    String singleNamespace = conf.get(Property.INSTANCE_DFS_DIR);
-    String ns = conf.get(Property.INSTANCE_VOLUMES);
-
-    String configuredBaseDirs[];
-
-    if (ns == null || ns.isEmpty()) {
-      configuredBaseDirs = new String[] {getDefaultBaseDir()};
-    } else {
-      String namespaces[] = ns.split(",");
-      String unescapedNamespaces[] = new String[namespaces.length];
-      int i = 0;
-      for (String namespace : namespaces) {
-        if (!namespace.contains(":")) {
-          throw new IllegalArgumentException("Expected fully qualified URI for " + Property.INSTANCE_VOLUMES.getKey()
+ " got " + namespace);
-        }
-
-        try {
-          // pass through URI to unescape hex encoded chars (e.g. convert %2C to "," char)
-          unescapedNamespaces[i++] = new Path(new URI(namespace)).toString();
-        } catch (URISyntaxException e) {
-          throw new IllegalArgumentException(Property.INSTANCE_VOLUMES.getKey() + " contains
" + namespace + " which has a syntax error", e);
-        }
-      }
-
-      configuredBaseDirs = prefix(unescapedNamespaces, singleNamespace);
-    }
-
-    return configuredBaseDirs;
-  }
-
   // these are functions to delay loading the Accumulo configuration unless we must
   public static synchronized String[] getBaseDirs() {
     if (baseDirs == null) {
-      baseDirs = checkBaseDirs(getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()),
false);
+      baseDirs = checkBaseDirs(VolumeConfiguration.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()),
false);
     }
 
     return baseDirs;
@@ -168,34 +116,24 @@ public class ServerConstants {
     return baseDirsList.toArray(new String[baseDirsList.size()]);
   }
 
-  public static String[] prefix(String bases[], String suffix) {
-    if (suffix.startsWith("/"))
-      suffix = suffix.substring(1);
-    String result[] = new String[bases.length];
-    for (int i = 0; i < bases.length; i++) {
-      result[i] = bases[i] + "/" + suffix;
-    }
-    return result;
-  }
-
   public static final String TABLE_DIR = "tables";
   public static final String RECOVERY_DIR = "recovery";
   public static final String WAL_DIR = "wal";
 
   public static String[] getTablesDirs() {
-    return prefix(getBaseDirs(), TABLE_DIR);
+    return VolumeConfiguration.prefix(getBaseDirs(), TABLE_DIR);
   }
 
   public static String[] getRecoveryDirs() {
-    return prefix(getBaseDirs(), RECOVERY_DIR);
+    return VolumeConfiguration.prefix(getBaseDirs(), RECOVERY_DIR);
   }
 
   public static String[] getWalDirs() {
-    return prefix(getBaseDirs(), WAL_DIR);
+    return VolumeConfiguration.prefix(getBaseDirs(), WAL_DIR);
   }
 
   public static String[] getWalogArchives() {
-    return prefix(getBaseDirs(), "walogArchive");
+    return VolumeConfiguration.prefix(getBaseDirs(), "walogArchive");
   }
 
   public static Path getInstanceIdLocation() {
@@ -209,11 +147,11 @@ public class ServerConstants {
   }
 
   public static String[] getMetadataTableDirs() {
-    return prefix(getTablesDirs(), MetadataTable.ID);
+    return VolumeConfiguration.prefix(getTablesDirs(), MetadataTable.ID);
   }
 
   public static String[] getTemporaryDirs() {
-    return prefix(getBaseDirs(), "tmp");
+    return VolumeConfiguration.prefix(getBaseDirs(), "tmp");
   }
 
   public static synchronized List<Pair<Path,Path>> getVolumeReplacements() {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
b/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
index 7f59eae..dc9acf8 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
@@ -51,7 +51,6 @@ import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.thrift.TKeyExtent;
 import org.apache.accumulo.core.file.FileOperations;
 import org.apache.accumulo.core.file.FileSKVIterator;
-import org.apache.accumulo.core.file.FileUtil;
 import org.apache.accumulo.core.metadata.MetadataTable;
 import org.apache.accumulo.core.security.Credentials;
 import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
@@ -64,6 +63,7 @@ import org.apache.accumulo.core.util.UtilWaitThread;
 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.server.util.FileUtil;
 import org.apache.accumulo.trace.instrument.TraceRunnable;
 import org.apache.accumulo.trace.instrument.Tracer;
 import org.apache.hadoop.conf.Configuration;
@@ -368,8 +368,7 @@ public class BulkImporter {
           Map<KeyExtent,Long> estimatedSizes = null;
           
           try {
-            FileSystem fs = vm.getFileSystemByPath(entry.getKey());
-            estimatedSizes = FileUtil.estimateSizes(acuConf, entry.getKey(), mapFileSizes.get(entry.getKey()),
extentsOf(entry.getValue()), conf, fs);
+            estimatedSizes = FileUtil.estimateSizes(acuConf, entry.getKey(), mapFileSizes.get(entry.getKey()),
extentsOf(entry.getValue()), conf, vm);
           } catch (IOException e) {
             log.warn("Failed to estimate map file sizes " + e.getMessage());
           }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
index 165c3b8..80301ef 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
@@ -34,7 +34,7 @@ import org.apache.accumulo.core.conf.DefaultConfiguration;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.KeyExtent;
-import org.apache.accumulo.core.file.FileUtil;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.util.CachedConfiguration;
 import org.apache.accumulo.server.ServerConstants;
 import org.apache.accumulo.server.client.HdfsZooInstance;
@@ -349,8 +349,8 @@ public class VolumeManagerImpl implements VolumeManager {
   public static VolumeManager get(AccumuloConfiguration conf) throws IOException {
     Map<String,FileSystem> fileSystems = new HashMap<String,FileSystem>();
     Configuration hadoopConf = CachedConfiguration.getInstance();
-    fileSystems.put(DEFAULT, FileUtil.getFileSystem(hadoopConf, conf));
-    for (String space : ServerConstants.getConfiguredBaseDirs(conf)) {
+    fileSystems.put(DEFAULT, VolumeConfiguration.getDefaultFilesystem(hadoopConf, conf));
+    for (String space : VolumeConfiguration.getConfiguredBaseDirs(conf)) {
       if (space.equals(DEFAULT))
         throw new IllegalArgumentException();
 

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java b/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
index 8533484..925f602 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
@@ -40,6 +40,7 @@ import org.apache.accumulo.core.data.KeyExtent;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.file.FileOperations;
 import org.apache.accumulo.core.file.FileSKVWriter;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.iterators.user.VersioningIterator;
 import org.apache.accumulo.core.master.state.tables.TableState;
 import org.apache.accumulo.core.master.thrift.MasterGoalState;
@@ -150,7 +151,7 @@ public class Initialize {
     else
       fsUri = FileSystem.getDefaultUri(conf).toString();
     log.info("Hadoop Filesystem is " + fsUri);
-    log.info("Accumulo data dirs are " + Arrays.asList(ServerConstants.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration())));
+    log.info("Accumulo data dirs are " + Arrays.asList(VolumeConfiguration.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration())));
     log.info("Zookeeper server is " + sconf.get(Property.INSTANCE_ZK_HOST));
     log.info("Checking if Zookeeper is available. If this hangs, then you need to make sure
zookeeper is running");
     if (!zookeeperAvailable()) {
@@ -172,7 +173,7 @@ public class Initialize {
     try {
       if (isInitialized(fs)) {
         String instanceDfsDir = sconf.get(Property.INSTANCE_DFS_DIR);
-        log.fatal("It appears the directories " + Arrays.asList(ServerConstants.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()))
+        log.fatal("It appears the directories " + Arrays.asList(VolumeConfiguration.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()))
             + " were previously initialized.");
         String instanceVolumes = sconf.get(Property.INSTANCE_VOLUMES);
         String instanceDfsUri = sconf.get(Property.INSTANCE_DFS_URI);
@@ -219,7 +220,7 @@ public class Initialize {
 
     UUID uuid = UUID.randomUUID();
     // the actual disk locations of the root table and tablets
-    String[] configuredTableDirs = ServerConstants.prefix(ServerConstants.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()),
+    String[] configuredTableDirs = VolumeConfiguration.prefix(VolumeConfiguration.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()),
         ServerConstants.TABLE_DIR);
     final Path rootTablet = new Path(fs.choose(configuredTableDirs) + "/" + RootTable.ID
+ RootTable.ROOT_TABLET_LOCATION);
     try {
@@ -295,13 +296,13 @@ public class Initialize {
   private static void initFileSystem(Opts opts, VolumeManager fs, UUID uuid, Path rootTablet)
throws IOException {
     FileStatus fstat;
 
-    initDirs(fs, uuid, ServerConstants.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()),
false);
+    initDirs(fs, uuid, VolumeConfiguration.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()),
false);
 
     // the actual disk locations of the metadata table and tablets
     final Path[] metadataTableDirs = paths(ServerConstants.getMetadataTableDirs());
 
-    String tableMetadataTabletDir = fs.choose(ServerConstants.prefix(ServerConstants.getMetadataTableDirs(),
TABLE_TABLETS_TABLET_DIR));
-    String defaultMetadataTabletDir = fs.choose(ServerConstants.prefix(ServerConstants.getMetadataTableDirs(),
Constants.DEFAULT_TABLET_LOCATION));
+    String tableMetadataTabletDir = fs.choose(VolumeConfiguration.prefix(ServerConstants.getMetadataTableDirs(),
TABLE_TABLETS_TABLET_DIR));
+    String defaultMetadataTabletDir = fs.choose(VolumeConfiguration.prefix(ServerConstants.getMetadataTableDirs(),
Constants.DEFAULT_TABLET_LOCATION));
 
     // initialize initial metadata config in zookeeper
     initMetadataConfig();
@@ -552,7 +553,7 @@ public class Initialize {
   }
 
   public static boolean isInitialized(VolumeManager fs) throws IOException {
-    for (String baseDir : ServerConstants.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()))
{
+    for (String baseDir : VolumeConfiguration.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()))
{
       if (fs.exists(new Path(baseDir, ServerConstants.INSTANCE_ID_DIR)) || fs.exists(new
Path(baseDir, ServerConstants.VERSION_DIR)))
         return true;
     }
@@ -563,10 +564,10 @@ public class Initialize {
   private static void addVolumes(VolumeManager fs) throws IOException {
     HashSet<String> initializedDirs = new HashSet<String>();
     initializedDirs
-        .addAll(Arrays.asList(ServerConstants.checkBaseDirs(ServerConstants.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()),
true)));
+        .addAll(Arrays.asList(ServerConstants.checkBaseDirs(VolumeConfiguration.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()),
true)));
 
     HashSet<String> uinitializedDirs = new HashSet<String>();
-    uinitializedDirs.addAll(Arrays.asList(ServerConstants.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration())));
+    uinitializedDirs.addAll(Arrays.asList(VolumeConfiguration.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration())));
     uinitializedDirs.removeAll(initializedDirs);
 
     Path aBasePath = new Path(initializedDirs.iterator().next());

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/DefaultServlet.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/DefaultServlet.java
b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/DefaultServlet.java
index 4aa2869..942f866 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/DefaultServlet.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/DefaultServlet.java
@@ -33,6 +33,7 @@ import javax.servlet.http.HttpServletResponse;
 
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.conf.DefaultConfiguration;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
 import org.apache.accumulo.core.util.Duration;
 import org.apache.accumulo.core.util.NumUtil;
@@ -41,7 +42,6 @@ import org.apache.accumulo.monitor.Monitor;
 import org.apache.accumulo.monitor.ZooKeeperStatus;
 import org.apache.accumulo.monitor.ZooKeeperStatus.ZooKeeperState;
 import org.apache.accumulo.monitor.util.celltypes.NumberType;
-import org.apache.accumulo.server.ServerConstants;
 import org.apache.accumulo.server.conf.ServerConfiguration;
 import org.apache.accumulo.server.fs.VolumeManager;
 import org.apache.accumulo.server.fs.VolumeManagerImpl;
@@ -268,7 +268,7 @@ public class DefaultServlet extends BasicServlet {
       long totalHdfsBytesUsed = 0l;
       
       try {
-        for (String baseDir : ServerConstants.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()))
{
+        for (String baseDir : VolumeConfiguration.getConfiguredBaseDirs(ServerConfiguration.getSiteConfiguration()))
{
           final Path basePath = new Path(baseDir);
           final FileSystem fs = vm.getFileSystemByPath(basePath);
           

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/server/tserver/src/main/java/org/apache/accumulo/tserver/BulkFailedCopyProcessor.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/BulkFailedCopyProcessor.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/BulkFailedCopyProcessor.java
index 891bea4..e9f1083 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/BulkFailedCopyProcessor.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/BulkFailedCopyProcessor.java
@@ -50,7 +50,7 @@ public class BulkFailedCopyProcessor implements Processor {
     Path tmp = new Path(dest.getParent(), dest.getName() + ".tmp");
     
     try {
-      FileSystem fs = TraceFileSystem.wrap(org.apache.accumulo.core.file.FileUtil.getFileSystem(CachedConfiguration.getInstance(),
+      FileSystem fs = TraceFileSystem.wrap(org.apache.accumulo.core.file.VolumeConfiguration.getDefaultFilesystem(CachedConfiguration.getInstance(),
           ServerConfiguration.getSiteConfiguration()));
       
       FileUtil.copy(fs, orig, fs, tmp, false, true, CachedConfiguration.getInstance());
@@ -58,7 +58,7 @@ public class BulkFailedCopyProcessor implements Processor {
       log.debug("copied " + orig + " to " + dest);
     } catch (IOException ex) {
       try {
-        FileSystem fs = TraceFileSystem.wrap(org.apache.accumulo.core.file.FileUtil.getFileSystem(CachedConfiguration.getInstance(),
+        FileSystem fs = TraceFileSystem.wrap(org.apache.accumulo.core.file.VolumeConfiguration.getDefaultFilesystem(CachedConfiguration.getInstance(),
             ServerConfiguration.getSiteConfiguration()));
         
         fs.create(dest).close();

http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8058a69/test/src/test/java/org/apache/accumulo/test/functional/BulkFileIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/functional/BulkFileIT.java b/test/src/test/java/org/apache/accumulo/test/functional/BulkFileIT.java
index 5d68155..c8023c0 100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/BulkFileIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/BulkFileIT.java
@@ -29,7 +29,7 @@ import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.file.FileOperations;
 import org.apache.accumulo.core.file.FileSKVWriter;
-import org.apache.accumulo.core.file.FileUtil;
+import org.apache.accumulo.core.file.VolumeConfiguration;
 import org.apache.accumulo.core.file.rfile.RFile;
 import org.apache.accumulo.core.security.Authorizations;
 import org.apache.accumulo.server.conf.ServerConfiguration;
@@ -53,7 +53,7 @@ public class BulkFileIT extends SimpleMacIT {
     c.tableOperations().addSplits(tableName, splits);
     Configuration conf = new Configuration();
     AccumuloConfiguration aconf = ServerConfiguration.getDefaultConfiguration();
-    FileSystem fs = TraceFileSystem.wrap(FileUtil.getFileSystem(conf, aconf));
+    FileSystem fs = TraceFileSystem.wrap(VolumeConfiguration.getDefaultFilesystem(conf, aconf));
 
     String dir = rootPath() + "/bulk_test_diff_files_89723987592_" + getTableNames(1)[0];
 


Mime
View raw message