MNEMONIC-218: Make inputSession & outputSession to accept the Configration parameter;

MNEMONIC-219: Remove the dependency on mapred FileOutputFormat.getUniqueFile

Author: paleyss

mnemonic-hadoop/mnemonic-hadoop-mapreduce/src/main/java/org/apache/mnemonic/hadoop/MneDurableInputSession.java

@@ -35,6 +35,7 @@ public class MneDurableInputSession<V>
     implements MneInputSession<V>, MneDurableComputable<NonVolatileMemAllocator> {
 
   private TaskAttemptContext taskAttemptContext;
+  private Configuration configuration;
   private String serviceName;
   private DurableType[] durableTypes;
   private EntityFactoryProxy[] entityFactoryProxies;
@@ -46,6 +47,11 @@ public class MneDurableInputSession<V>
 
   public MneDurableInputSession(TaskAttemptContext taskAttemptContext) {
     setTaskAttemptContext(taskAttemptContext);
+    setConfiguration(taskAttemptContext.getConfiguration());
+  }
+
+  public MneDurableInputSession(Configuration configuration) {
+    setConfiguration(configuration);
   }
 
   public void validateConfig() {
@@ -61,10 +67,10 @@ && getEntityFactoryProxies().length < 1) { /* T.B.D. BUFFER & CHUNK */
 
   @Override
   public void readConfig(String prefix) {
-    if (getTaskAttemptContext() == null) {
-      throw new ConfigurationException("taskAttemptContext has not yet been set");
+    if (getConfiguration() == null) {
+      throw new ConfigurationException("configuration has not yet been set");
     }
-    Configuration conf = getTaskAttemptContext().getConfiguration();
+    Configuration conf = getConfiguration();
     setServiceName(MneConfigHelper.getMemServiceName(conf, MneConfigHelper.DEFAULT_INPUT_CONFIG_PREFIX));
     setDurableTypes(MneConfigHelper.getDurableTypes(conf, MneConfigHelper.DEFAULT_INPUT_CONFIG_PREFIX));
     setEntityFactoryProxies(Utils.instantiateEntityFactoryProxies(
@@ -144,4 +150,12 @@ public NonVolatileMemAllocator getAllocator() {
   public long getHandler() {
     return m_handler;
   }
+
+  public Configuration getConfiguration() {
+    return configuration;
+  }
+
+  public void setConfiguration(Configuration configuration) {
+    this.configuration = configuration;
+  }
 }

mnemonic-hadoop/mnemonic-hadoop-mapreduce/src/main/java/org/apache/mnemonic/hadoop/MneDurableOutputSession.java

@@ -28,21 +28,25 @@
 import org.apache.mnemonic.collections.DurableSinglyLinkedList;
 import org.apache.mnemonic.collections.DurableSinglyLinkedListFactory;
 
+import java.text.NumberFormat;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
 
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.JobContext;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.TaskID;
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 
 public class MneDurableOutputSession<V>
     implements MneOutputSession<V>, MneDurableComputable<NonVolatileMemAllocator> {
 
   private long poolSize;
   private TaskAttemptContext taskAttemptContext;
+  private Configuration configuration;
   private String serviceName;
   private DurableType[] durableTypes;
   private EntityFactoryProxy[] entityFactoryProxies;
@@ -61,6 +65,11 @@ public class MneDurableOutputSession<V>
   public MneDurableOutputSession(TaskAttemptContext taskAttemptContext) {
     setTaskAttemptContext(taskAttemptContext);
     m_recordmap = new HashMap<V, DurableSinglyLinkedList<V>>();
+    setConfiguration(taskAttemptContext.getConfiguration());
+  }
+  
+  public MneDurableOutputSession(Configuration configuration) {
+    setConfiguration(configuration);
   }
 
   public void validateConfig() {
@@ -79,7 +88,7 @@ public void readConfig(String prefix) {
     if (getTaskAttemptContext() == null) {
       throw new ConfigurationException("taskAttemptContext has not yet been set");
     }
-    Configuration conf = getTaskAttemptContext().getConfiguration();
+    Configuration conf = getConfiguration();
     setServiceName(MneConfigHelper.getMemServiceName(conf, MneConfigHelper.DEFAULT_OUTPUT_CONFIG_PREFIX));
     setDurableTypes(MneConfigHelper.getDurableTypes(conf, MneConfigHelper.DEFAULT_OUTPUT_CONFIG_PREFIX));
     setEntityFactoryProxies(Utils.instantiateEntityFactoryProxies(
@@ -93,10 +102,40 @@ public void readConfig(String prefix) {
 
   protected Path genNextPoolPath() {
     Path ret = new Path(FileOutputFormat.getOutputPath(getTaskAttemptContext()),
-        FileOutputFormat.getUniqueFile(getTaskAttemptContext(),
-            String.format("%s-%05d", getBaseOutputName(), ++m_poolidx), MneConfigHelper.DEFAULT_FILE_EXTENSION));
+        getUniqueName(String.format("%s-%05d", getBaseOutputName(), ++m_poolidx), 
+            MneConfigHelper.DEFAULT_FILE_EXTENSION));
     return ret;
   }
+  
+  protected String getUniqueName(String name, String extension) {
+    int partition;
+    
+    NumberFormat numberFormat = NumberFormat.getInstance();
+    numberFormat.setMinimumIntegerDigits(5);
+    numberFormat.setGroupingUsed(false);
+    
+    if (null != getTaskAttemptContext()) {
+      TaskID taskId = getTaskAttemptContext().getTaskAttemptID().getTaskID();
+      partition = taskId.getId();
+    } else {
+      partition = getConfiguration().getInt(JobContext.TASK_PARTITION, -1);
+    } 
+    if (partition == -1) {
+      throw new IllegalArgumentException("This method can only be called from an application");
+    }
+    
+    String taskType = getConfiguration().getBoolean(JobContext.TASK_ISMAP, JobContext.DEFAULT_TASK_ISMAP) ? "m" : "r";
+    
+    StringBuilder result = new StringBuilder();
+    result.append(name);
+    result.append('-');
+    result.append(taskType);
+    result.append('-');
+    result.append(numberFormat.format(partition));
+    result.append(extension);
+    return result.toString();
+    
+  }
 
   @Override
   public void initNextPool() {
@@ -326,4 +365,12 @@ public void setBaseOutputName(String baseOutputName) {
     this.baseOutputName = baseOutputName;
   }
 
+  public Configuration getConfiguration() {
+    return configuration;
+  }
+
+  public void setConfiguration(Configuration configuration) {
+    this.configuration = configuration;
+  }
+
 }