diff --git a/CHANGES.txt b/CHANGES.txt
index c097e19cfd..9979aefe66 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -471,6 +471,9 @@ Trunk (unreleased changes)
     HADOOP-2366. Support trimmed strings in Configuration.  (Michele Catasta
     via szetszwo)
 
+    HADOOP-6099. The RPC module can be configured to not send period pings.
+    The default behaviour of sending periodic pings remain unchanged. (dhruba)
+
   OPTIMIZATIONS
 
     HADOOP-5595. NameNode does not need to run a replicator to choose a
diff --git a/src/java/org/apache/hadoop/ipc/Client.java b/src/java/org/apache/hadoop/ipc/Client.java
index 1d01faf673..f71a4a507a 100644
--- a/src/java/org/apache/hadoop/ipc/Client.java
+++ b/src/java/org/apache/hadoop/ipc/Client.java
@@ -73,6 +73,7 @@ public class Client {
   final private int maxRetries; //the max. no. of retries for socket connections
   private boolean tcpNoDelay; // if T then disable Nagle's Algorithm
   private int pingInterval; // how often sends ping to the server in msecs
+  final private boolean doPing; //do we need to send ping message
 
   private SocketFactory socketFactory;           // how to create sockets
   private int refCount = 1;
@@ -101,6 +102,22 @@ final public static void setPingInterval(Configuration conf, int pingInterval) {
   final static int getPingInterval(Configuration conf) {
     return conf.getInt(PING_INTERVAL_NAME, DEFAULT_PING_INTERVAL);
   }
+
+  /**
+   * The time after which a RPC will timeout.
+   * If ping is not enabled (via ipc.client.ping), then the timeout value is the 
+   * same as the pingInterval.
+   * If ping is enabled, then there is no timeout value.
+   * 
+   * @param conf Configuration
+   * @return the timeout period in milliseconds. -1 if no timeout value is set
+   */
+  final public static int getTimeout(Configuration conf) {
+    if (!conf.getBoolean("ipc.client.ping", true)) {
+      return getPingInterval(conf);
+    }
+    return -1;
+  }
   
   /**
    * Increment this client's reference count
@@ -317,8 +334,13 @@ private synchronized void setupIOstreams() {
             handleConnectionFailure(ioFailures++, maxRetries, ie);
           }
         }
-        this.in = new DataInputStream(new BufferedInputStream
+        if (doPing) {
+          this.in = new DataInputStream(new BufferedInputStream
             (new PingInputStream(NetUtils.getInputStream(socket))));
+        } else {
+          this.in = new DataInputStream(new BufferedInputStream
+            (NetUtils.getInputStream(socket)));
+        }
         this.out = new DataOutputStream
             (new BufferedOutputStream(NetUtils.getOutputStream(socket)));
         writeHeader();
@@ -634,6 +656,7 @@ public Client(Class<? extends Writable> valueClass, Configuration conf,
       conf.getInt("ipc.client.connection.maxidletime", 10000); //10s
     this.maxRetries = conf.getInt("ipc.client.connect.max.retries", 10);
     this.tcpNoDelay = conf.getBoolean("ipc.client.tcpnodelay", false);
+    this.doPing = conf.getBoolean("ipc.client.ping", true);
     this.pingInterval = getPingInterval(conf);
     if (LOG.isDebugEnabled()) {
       LOG.debug("The ping interval is" + this.pingInterval + "ms.");
diff --git a/src/test/core/org/apache/hadoop/ipc/TestRPC.java b/src/test/core/org/apache/hadoop/ipc/TestRPC.java
index d0db263cc1..e12cfaab73 100644
--- a/src/test/core/org/apache/hadoop/ipc/TestRPC.java
+++ b/src/test/core/org/apache/hadoop/ipc/TestRPC.java
@@ -231,7 +231,7 @@ public void testSlowRpc() throws Exception {
   }
 
 
-  public void testCalls() throws Exception {
+  public void testCalls(Configuration conf) throws Exception {
     Server server = RPC.getServer(new TestImpl(), ADDRESS, 0, conf);
     TestProtocol proxy = null;
     try {
@@ -382,10 +382,20 @@ public void testAuthorization() throws Exception {
     conf.set(ACL_CONFIG, "invalid invalid");
     doRPCs(conf, true);
   }
+
+  /**
+   * Switch off setting socketTimeout values on RPC sockets.
+   * Verify that RPC calls still work ok.
+   */
+  public void testNoPings() throws Exception {
+    Configuration conf = new Configuration();
+    conf.setBoolean("ipc.client.ping", false);
+    new TestRPC("testnoPings").testCalls(conf);
+  }
   
   public static void main(String[] args) throws Exception {
 
-    new TestRPC("test").testCalls();
+    new TestRPC("test").testCalls(conf);
 
   }
 }