<commit_before>diff --git a/libhdfs/hdfsJniHelper.c b/libhdfs/hdfsJniHelper.c
index f7bf5e4..d7fbd8d 100644
--- a/libhdfs/hdfsJniHelper.c
+++ b/libhdfs/hdfsJniHelper.c
@@ -15,7 +15,6 @@
  */
 
 #include <string.h> 
-#include <error.h>
 #include "hdfsJniHelper.h"
 
 static pthread_mutex_t hdfsHashMutex = PTHREAD_MUTEX_INITIALIZER;
diff --git a/org/apache/hadoop/mapred/pipes/Application.java b/org/apache/hadoop/mapred/pipes/Application.java
index 8d4d259..060ed26 100644
--- a/org/apache/hadoop/mapred/pipes/Application.java
+++ b/org/apache/hadoop/mapred/pipes/Application.java
@@ -47,6 +47,7 @@ import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.mapred.TaskAttemptID;
 import org.apache.hadoop.mapred.TaskLog;
+import org.apache.hadoop.mapred.TaskTracker;
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.filecache.DistributedCache;
 import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
@@ -96,14 +97,22 @@ class Application<K1 extends WritableComparable, V1 extends Writable,
     env.put(Submitter.PORT, 
             Integer.toString(serverSocket.getLocalPort()));
     
+    TaskAttemptID taskid = 
+      TaskAttemptID.forName(conf.get(MRJobConfig.TASK_ATTEMPT_ID));
+
+    // get the task's working directory
+    String workDir = TaskTracker.getLocalTaskDir(conf.getUser(),
+            taskid.getJobID().toString(),
+            taskid.getTaskID().toString());
+
     //Add token to the environment if security is enabled
     Token<JobTokenIdentifier> jobToken = TokenCache.getJobToken(conf
         .getCredentials());
     // This password is used as shared secret key between this application and
     // child pipes process
     byte[]  password = jobToken.getPassword();
-    String localPasswordFile = new File(".") + Path.SEPARATOR
-        + "jobTokenPassword";
+
+    String localPasswordFile = new File(workDir, "jobTokenPassword").getAbsolutePath();
     writePasswordToLocalFile(localPasswordFile, password, conf);
     env.put("hadoop.pipes.shared.secret.location", localPasswordFile);
  
@@ -122,8 +131,6 @@ class Application<K1 extends WritableComparable, V1 extends Writable,
     // wrap the command in a stdout/stderr capture
     // we are starting map/reduce task of the pipes job. this is not a cleanup
     // attempt. 
-    TaskAttemptID taskid = 
-      TaskAttemptID.forName(conf.get(MRJobConfig.TASK_ATTEMPT_ID));
     File stdout = TaskLog.getTaskLogFile(taskid, false, TaskLog.LogName.STDOUT);
     File stderr = TaskLog.getTaskLogFile(taskid, false, TaskLog.LogName.STDERR);
     long logLength = TaskLog.getTaskLogLength(conf);
diff --git a/pipes/impl/HadoopPipes.cc b/pipes/impl/HadoopPipes.cc
index 1d2fc95..c5e2a95 100644
--- a/pipes/impl/HadoopPipes.cc
+++ b/pipes/impl/HadoopPipes.cc
@@ -127,7 +127,7 @@ namespace HadoopPipes {
     static const char lineSeparator = '\n';
 
     void writeBuffer(const string& buffer) {
-      fprintf(stream, quoteString(buffer, "\t\n").c_str());
+      fprintf(stream, "%s", quoteString(buffer, "\t\n").c_str());
     }
 
   public:
@@ -803,14 +803,20 @@ namespace HadoopPipes {
       if (numReduces != 0) { 
         reducer = factory->createCombiner(*this);
         partitioner = factory->createPartitioner(*this);
-      }
-      if (reducer != NULL) {
-        int64_t spillSize = 100;
-        if (jobConf->hasKey("mapreduce.task.io.sort.mb")) {
-          spillSize = jobConf->getInt("mapreduce.task.io.sort.mb");
-        }
-        writer = new CombineRunner(spillSize * 1024 * 1024, this, reducer, 
-                                   uplink, partitioner, numReduces);
+	if (reducer != NULL) {
+	  int64_t spillSize = 100;
+	  if (jobConf->hasKey("mapreduce.task.io.sort.mb")) {
+	    spillSize = jobConf->getInt("mapreduce.task.io.sort.mb");
+	  }
+	  writer = new CombineRunner(spillSize * 1024 * 1024, this, reducer, 
+				     uplink, partitioner, numReduces);
+	} 
+      } else {
+	if (jobConf->hasKey("hadoop.pipes.java.recordwriter")) {
+	  if (jobConf->get("hadoop.pipes.java.recordwriter") == "false") {
+	    writer = factory->createRecordWriter(*this);
+	  }
+	}
       }
       hasTask = true;
     }
diff --git a/utils/impl/SerialUtils.cc b/utils/impl/SerialUtils.cc
index 03d009b..3736373 100644
--- a/utils/impl/SerialUtils.cc
+++ b/utils/impl/SerialUtils.cc
@@ -19,6 +19,7 @@
 #include "hadoop/StringUtils.hh"
 
 #include <errno.h>
+#include <stdint.h>
 #include <rpc/types.h>
 #include <rpc/xdr.h>
 #include <string>
@@ -252,13 +253,15 @@ namespace HadoopUtils {
     stream.write(buf, sizeof(float));
   }
 
-  void deserializeFloat(float& t, InStream& stream)
+  float deserializeFloat(InStream& stream)
   {
+    float t;
     char buf[sizeof(float)];
     stream.read(buf, sizeof(float));
     XDR xdrs;
     xdrmem_create(&xdrs, buf, sizeof(float), XDR_DECODE);
     xdr_float(&xdrs, &t);<commit_msg>updated patch for cdh4.3 (fixed env problem)<commit_after>diff --git a/libhdfs/hdfsJniHelper.c b/libhdfs/hdfsJniHelper.c
index f7bf5e4..d7fbd8d 100644
--- a/libhdfs/hdfsJniHelper.c
+++ b/libhdfs/hdfsJniHelper.c
@@ -15,7 +15,6 @@
  */
 
 #include <string.h> 
-#include <error.h>
 #include "hdfsJniHelper.h"
 
 static pthread_mutex_t hdfsHashMutex = PTHREAD_MUTEX_INITIALIZER;
diff --git a/org/apache/hadoop/mapred/pipes/Application.java b/org/apache/hadoop/mapred/pipes/Application.java
index 8d4d259..060ed26 100644
--- a/org/apache/hadoop/mapred/pipes/Application.java
+++ b/org/apache/hadoop/mapred/pipes/Application.java
@@ -47,6 +47,7 @@ import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.mapred.TaskAttemptID;
 import org.apache.hadoop.mapred.TaskLog;
+import org.apache.hadoop.mapred.TaskTracker;
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.filecache.DistributedCache;
 import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
@@ -96,14 +97,22 @@ class Application<K1 extends WritableComparable, V1 extends Writable,
     env.put(Submitter.PORT, 
             Integer.toString(serverSocket.getLocalPort()));
     
+    TaskAttemptID taskid = 
+      TaskAttemptID.forName(conf.get(MRJobConfig.TASK_ATTEMPT_ID));
+
+    // get the task's working directory
+    String workDir = TaskTracker.getLocalTaskDir(conf.getUser(),
+            taskid.getJobID().toString(),
+            taskid.getTaskID().toString());
+
     //Add token to the environment if security is enabled
     Token<JobTokenIdentifier> jobToken = TokenCache.getJobToken(conf
         .getCredentials());
     // This password is used as shared secret key between this application and
     // child pipes process
     byte[]  password = <PASSWORD>();
-    String localPasswordFile = new File(".") + Path.SEPARATOR
-        + "jobTokenPassword";
+
+    String localPasswordFile = new File(workDir, "jobTokenPassword").getAbsolutePath();
     writePasswordToLocalFile(localPasswordFile, password, conf);
     env.put("hadoop.pipes.shared.secret.location", localPasswordFile);
  
@@ -122,8 +131,6 @@ class Application<K1 extends WritableComparable, V1 extends Writable,
     // wrap the command in a stdout/stderr capture
     // we are starting map/reduce task of the pipes job. this is not a cleanup
     // attempt. 
-    TaskAttemptID taskid = 
-      TaskAttemptID.forName(conf.get(MRJobConfig.TASK_ATTEMPT_ID));
     File stdout = TaskLog.getTaskLogFile(taskid, false, TaskLog.LogName.STDOUT);
     File stderr = TaskLog.getTaskLogFile(taskid, false, TaskLog.LogName.STDERR);
     long logLength = TaskLog.getTaskLogLength(conf);
diff --git a/pipes/impl/HadoopPipes.cc b/pipes/impl/HadoopPipes.cc
index 1d2fc95..fdb50c2 100644
--- a/pipes/impl/HadoopPipes.cc
+++ b/pipes/impl/HadoopPipes.cc
@@ -83,6 +83,13 @@ namespace HadoopPipes {
     }
   };
 
+char* _get_command_port(){
+	char* mapreduce_command_port = getenv("mapreduce.pipes.command.port");
+  if(mapreduce_command_port != NULL)
+    return mapreduce_command_port;
+  return getenv("hadoop.pipes.command.port");
+  }
+
   class DownwardProtocol {
   public:
     virtual void start(int protocol) = 0;
@@ -127,7 +134,7 @@ namespace HadoopPipes {
     static const char lineSeparator = '\n';
 
     void writeBuffer(const string& buffer) {
-      fprintf(stream, quoteString(buffer, "\t\n").c_str());
+      fprintf(stream, "%s", quoteString(buffer, "\t\n").c_str());
     }
 
   public:
@@ -803,14 +810,20 @@ namespace HadoopPipes {
       if (numReduces != 0) { 
         reducer = factory->createCombiner(*this);
         partitioner = factory->createPartitioner(*this);
-      }
-      if (reducer != NULL) {
-        int64_t spillSize = 100;
-        if (jobConf->hasKey("mapreduce.task.io.sort.mb")) {
-          spillSize = jobConf->getInt("mapreduce.task.io.sort.mb");
-        }
-        writer = new CombineRunner(spillSize * 1024 * 1024, this, reducer, 
-                                   uplink, partitioner, numReduces);
+	if (reducer != NULL) {
+	  int64_t spillSize = 100;
+	  if (jobConf->hasKey("mapreduce.task.io.sort.mb")) {
+	    spillSize = jobConf->getInt("mapreduce.task.io.sort.mb");
+	  }
+	  writer = new CombineRunner(spillSize * 1024 * 1024, this, reducer, 
+				     uplink, partitioner, numReduces);
+	} 
+      } else {
+	if (jobConf->hasKey("hadoop.pipes.java.recordwriter")) {
+	  if (jobConf->get("hadoop.pipes.java.recordwriter") == "false") {
+	    writer = factory->createRecordWriter(*this);
+	  }
+	}
       }
       hasTask = true;
     }
@@ -1042,7 +1055,7 @@ namespace HadoopPipes {
    */
   void* ping(void* ptr) {
     TaskContextImpl* context = (TaskContextImpl*) ptr;
-    char* portStr = getenv("mapreduce.pipes.command.port");
+    char* portStr = _get_command_port();
     int MAX_RETRIES = 3;
     int remaining_retries = MAX_RETRIES;
     while (!context->isDone()) {
@@ -1095,7 +1108,7 @@ namespace HadoopPipes {
     try {
       TaskContextImpl* context = new TaskContextImpl(factory);
       Protocol* connection;
-      char* portStr = getenv("mapreduce.pipes.command.port");
+      char* portStr = _get_command_port();
       int sock = -1;
       FILE* stream = NULL;
       FILE* outStream = NULL;
diff --git a/utils/impl/SerialUtils.cc b/utils/impl/SerialUtils.cc
index 03d009b..3736373 100644
--- a/utils/impl/SerialUtils.cc
+++ b/utils/impl/SerialUtils.cc
@@ -19,6 +19,7 @@
 #include "hadoop/StringUtils.hh"
 
 #include <errno.h>
+#include <stdint.h>
 #include <rpc/types.h>
 #include <rpc/xdr.h>
 #include <string>
@@ -252,13 +253,15 @@ namespace HadoopUtils {
     stream.write(buf, sizeof(float));
   }
 
-  void deserializeFloat(float& t, InStream& stream)
+  float deserializeFloat(InStream& stream)
   {
+    float t;
     char buf[sizeof(float)];
     stream.read(buf, sizeof(float));
     XDR xdrs;
     xdrmem_create(&xdrs, buf, sizeof(float), XDR_DECODE);
     xdr_float(&xdrs, &t);
{'fraction_non_alphanumeric': 0.14304017571154024, 'fraction_numerical': 0.030658003111558526, 'mean_word_length': 3.388755020080321, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5381684', 'n_tokens_mistral': 4033, 'n_tokens_neox': 3738, 'n_words': 695}