diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
index 567f49e..1838576 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
@@ -210,7 +210,7 @@
    */
 
   public static class BatchMutateContext {
-      private BatchMutatePhase currentPhase = BatchMutatePhase.PRE;
+      private volatile BatchMutatePhase currentPhase = BatchMutatePhase.PRE;
       // The max of reference counts on the pending rows of this batch at the time this batch arrives
       private int maxPendingRowCount = 0;
       private final int clientVersion;
@@ -273,12 +273,24 @@
       }
 
       public CountDownLatch getCountDownLatch() {
-          if (waitList == null) {
-              waitList = new ArrayList<>();
+          synchronized (this) {
+              if (waitList == null) {
+                  waitList = new ArrayList<>();
+              }
+              CountDownLatch countDownLatch = new CountDownLatch(1);
+              waitList.add(countDownLatch);
+              return countDownLatch;
           }
-          CountDownLatch countDownLatch = new CountDownLatch(1);
-          waitList.add(countDownLatch);
-          return countDownLatch;
+      }
+
+      public void countDownAllLatches() {
+          synchronized (this) {
+              if (waitList != null) {
+                  for (CountDownLatch countDownLatch : waitList) {
+                      countDownLatch.countDown();
+                  }
+              }
+          }
       }
 
       public int getMaxPendingRowCount() {
@@ -1067,11 +1079,9 @@
 
     private void waitForPreviousConcurrentBatch(TableName table, BatchMutateContext context)
             throws Throwable {
-        boolean done;
-        BatchMutatePhase phase;
-        done = true;
+        boolean done = true;
         for (BatchMutateContext lastContext : context.lastConcurrentBatchContext.values()) {
-            phase = lastContext.getCurrentPhase();
+            BatchMutatePhase phase = lastContext.getCurrentPhase();
 
             if (phase == BatchMutatePhase.PRE) {
                 CountDownLatch countDownLatch = lastContext.getCountDownLatch();
@@ -1191,7 +1201,7 @@
             // Release the locks before making RPC calls for index updates
             unlockRows(context);
             // Do the first phase index updates
-            doPre(c, context, miniBatchOp);
+            doPre(context);
             // Acquire the locks again before letting the region proceed with data table updates
             lockRows(context);
             if (context.lastConcurrentBatchContext != null) {
@@ -1277,9 +1287,13 @@
     }
 
     /**
-     * When this hook is called, all the rows in the batch context are locked. Because the rows
-     * are locked, we can safely make updates to the context object and perform the necessary
-     * cleanup.
+     * When this hook is called, all the rows in the batch context are locked if the batch of
+     * mutations is successful. Because the rows are locked, we can safely make updates to
+     * pending row states in memory and perform the necessary cleanup in that case.
+     *
+     * However, when the batch fails, then some of the rows may not be locked. In that case,
+     * we remove the pending row states from the concurrent hash map without updating them since
+     * pending rows states become invalid when a batch fails.
      */
   @Override
   public void postBatchMutateIndispensably(ObserverContext<RegionCoprocessorEnvironment> c,
@@ -1297,11 +1311,7 @@
           } else {
               context.currentPhase = BatchMutatePhase.FAILED;
           }
-          if (context.waitList != null) {
-              for (CountDownLatch countDownLatch : context.waitList) {
-                  countDownLatch.countDown();
-              }
-          }
+          context.countDownAllLatches();
           removePendingRows(context);
           if (context.indexUpdates != null) {
               context.indexUpdates.clear();
@@ -1361,6 +1371,16 @@
   }
 
   private void removePendingRows(BatchMutateContext context) {
+      if (context.currentPhase == BatchMutatePhase.FAILED) {
+          // This batch failed. All concurrent batches will fail too. So we can remove
+          // all rows of this batch from the memory as the in-memory row images are not valid
+          // anymore. Please note that when a batch fails, some of the rows may not have been
+          // locked and so it is not safe to update the pending row entries in that case.
+          for (ImmutableBytesPtr rowKey : context.rowsToLock) {
+              pendingRows.remove(rowKey);
+          }
+          return;
+      }
       for (RowLock rowLock : context.rowLocks) {
           ImmutableBytesPtr rowKey = rowLock.getRowKey();
           PendingRow pendingRow = pendingRows.get(rowKey);
@@ -1373,10 +1393,10 @@
       }
   }
 
-  private void doPre(ObserverContext<RegionCoprocessorEnvironment> c, BatchMutateContext context,
-                     MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
-      long start = EnvironmentEdgeManager.currentTimeMillis();
+  private void doPre(BatchMutateContext context) throws IOException {
+      long start = 0;
       try {
+          start = EnvironmentEdgeManager.currentTimeMillis();
           if (failPreIndexUpdatesForTesting) {
               throw new DoNotRetryIOException("Simulating the first (i.e., pre) index table write failure");
           }
@@ -1394,8 +1414,6 @@
           lockRows(context);
           rethrowIndexingException(e);
       }
-      throw new RuntimeException(
-              "Somehow didn't complete the index update, but didn't return succesfully either!");
   }
 
   private void extractExpressionsAndColumns(DataInputStream input,
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java
index f02018f..4bf5ffa 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java
@@ -89,6 +89,10 @@
         props.put(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB, Long.toString(0));
         props.put(BaseScannerRegionObserverConstants.PHOENIX_MAX_LOOKBACK_AGE_CONF_KEY,
             Integer.toString(MAX_LOOKBACK_AGE));
+        // The following sets the row lock wait duration to 10 ms to test the code path handling
+        // row lock timeouts. When there are concurrent mutations, the wait time can be
+        // much longer than 10 ms.
+        props.put("hbase.rowlock.wait.duration", "10");
         setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
     }
     @Parameterized.Parameters(
@@ -300,9 +304,9 @@
 
     @Test
     public void testConcurrentUpserts() throws Exception {
-        int nThreads = 4;
-        final int batchSize = 200;
-        final int nRows = 51;
+        int nThreads = 10;
+        final int batchSize = 20;
+        final int nRows = 100;
         final int nIndexValues = 23;
         final String tableName = generateUniqueName();
         final String indexName = generateUniqueName();
@@ -333,6 +337,7 @@
                         }
                         conn.commit();
                     } catch (SQLException e) {
+                        System.out.println(e);
                         throw new RuntimeException(e);
                     } finally {
                         doneSignal.countDown();