svn commit: r1235460 - in /incubator/accumulo/branches/1.4/src: core/src/main/java/org/apache/accumulo/core/conf/Property.java server/src/main/java/org/apache/accumulo/server/logger/LogArchiver.java

[kt...@apache.org]

Author: kturner
Date: Tue Jan 24 20:10:32 2012
New Revision: 1235460

URL: http://svn.apache.org/viewvc?rev=1235460&view=rev
Log:
ACCUMULO-326 Made walog archive replication factor configurable

Modified:
    incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/conf/Property.java
    incubator/accumulo/branches/1.4/src/server/src/main/java/org/apache/accumulo/server/logger/LogArchiver.java

Modified: incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/conf/Property.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/conf/Property.java?rev=1235460&r1=1235459&r2=1235460&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/conf/Property.java (original)
+++ incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/conf/Property.java Tue Jan 24 20:10:32 2012
@@ -157,6 +157,8 @@ public enum Property {
       "The directory used to store write-ahead logs on the local filesystem. It is possible to specify a comma-separated list of directories."),
   LOGGER_PORTSEARCH("logger.port.search", "false", PropertyType.BOOLEAN, "if the port above is in use, search higher ports until one is available"),
   LOGGER_ARCHIVE("logger.archive", "false", PropertyType.BOOLEAN, "determines if logs are archived in hdfs"),
+  LOGGER_ARCHIVE_REPLICATION("logger.archive.replication", "0", PropertyType.COUNT,
+      "determines the replication factor for walogs archived in hdfs, set to zero to use default"),
   LOGGER_MONITOR_FS(
       "logger.monitor.fs",
       "true",

Modified: incubator/accumulo/branches/1.4/src/server/src/main/java/org/apache/accumulo/server/logger/LogArchiver.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/server/src/main/java/org/apache/accumulo/server/logger/LogArchiver.java?rev=1235460&r1=1235459&r2=1235460&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/server/src/main/java/org/apache/accumulo/server/logger/LogArchiver.java (original)
+++ incubator/accumulo/branches/1.4/src/server/src/main/java/org/apache/accumulo/server/logger/LogArchiver.java Tue Jan 24 20:10:32 2012
@@ -43,6 +43,7 @@ public class LogArchiver {
   private FileSystem dest;
   private String destDir;
   private final boolean archive;
+  private short replication;
   
   static Path archiveName(String fullPath) {
     if (isArchive(fullPath))
@@ -82,7 +83,8 @@ public class LogArchiver {
         dest.delete(new Path(destDir, name + ".gz_tmp"), false);
         
         FSDataInputStream in = src.open(srcPath);
-        GZIPOutputStream out = new GZIPOutputStream(dest.create(new Path(destDir, name + ".gz_tmp"), false));
+        GZIPOutputStream out = new GZIPOutputStream(dest.create(new Path(destDir, name + ".gz_tmp"), false, dest.getConf().getInt("io.file.buffer.size", 4096),
+            replication == 0 ? dest.getDefaultReplication() : replication, dest.getDefaultBlockSize()));
         
         byte buf[] = new byte[1 << 20];
         int numRead;
@@ -112,6 +114,7 @@ public class LogArchiver {
     this.threadPool = Executors.newSingleThreadExecutor();
     this.destDir = conf.get(Property.INSTANCE_DFS_DIR) + "/walogArchive";
     this.archive = archive;
+    this.replication = (short) conf.getCount(Property.LOGGER_ARCHIVE_REPLICATION);
     dest.mkdirs(new Path(destDir));
     
     if (archive) {