From 1ae64ccee03d587bfeae9c84b946a3189136df1d Mon Sep 17 00:00:00 2001
From: Michael Stack <stack@apache.org>
Date: Mon, 12 Feb 2018 10:47:06 -0800
Subject: [PATCH] HBASE-19972 Should rethrow the
 RetriesExhaustedWithDetailsException when failed to apply the batch in
 ReplicationSink; ADDENDUM to fix TestReplicationSink

---
 .../hbase/master/assignment/AssignProcedure.java    |  4 ++--
 .../regionserver/TestReplicationSink.java           | 13 +++++++------
 2 files changed, 9 insertions(+), 8 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignProcedure.java
index 801caf5eacc..65eafe72cea 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignProcedure.java
@@ -158,7 +158,7 @@ public class AssignProcedure extends RegionTransitionProcedure {
       LOG.info("Assigned, not reassigning; " + this + "; " + regionNode.toShortString());
       return false;
     }
-    // Don't assign if table is in disabling of disabled state.
+    // Don't assign if table is in disabling or disabled state.
     TableStateManager tsm = env.getMasterServices().getTableStateManager();
     TableName tn = regionNode.getRegionInfo().getTable();
     if (tsm.isTableState(tn, TableState.State.DISABLING, TableState.State.DISABLED)) {
@@ -166,7 +166,7 @@ public class AssignProcedure extends RegionTransitionProcedure {
       return false;
     }
     // If the region is SPLIT, we can't assign it. But state might be CLOSED, rather than
-    // SPLIT which is what a region gets set to when Unassigned as part of SPLIT. FIX.
+    // SPLIT which is what a region gets set to when unassigned as part of SPLIT. FIX.
     if (regionNode.isInState(State.SPLIT) ||
         (regionNode.getRegionInfo().isOffline() && regionNode.getRegionInfo().isSplit())) {
       LOG.info("SPLIT, cannot be assigned; " + this + "; " + regionNode +
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSink.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSink.java
index fcce84fa1a0..aa6c39cd53a 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSink.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSink.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -317,8 +317,8 @@ public class TestReplicationSink {
     Path dir = TEST_UTIL.getDataTestDirOnTestFS("testReplicateEntries");
     Path familyDir = new Path(dir, Bytes.toString(FAM_NAME1));
     int numRows = 10;
-
     List<Path> p = new ArrayList<>(1);
+    final String hfilePrefix = "hfile-";
 
     // 1. Generate 25 hfile ranges
     Random rng = new SecureRandom();
@@ -335,7 +335,7 @@ public class TestReplicationSink {
     FileSystem fs = dir.getFileSystem(conf);
     Iterator<Integer> numbersItr = numberList.iterator();
     for (int i = 0; i < 25; i++) {
-      Path hfilePath = new Path(familyDir, "hfile_" + i);
+      Path hfilePath = new Path(familyDir, hfilePrefix + i);
       HFileTestUtil.createHFile(conf, fs, hfilePath, FAM_NAME1, FAM_NAME1,
         Bytes.toBytes(numbersItr.next()), Bytes.toBytes(numbersItr.next()), numRows);
       p.add(hfilePath);
@@ -370,10 +370,10 @@ public class TestReplicationSink {
               .append(Bytes.toString(TABLE_NAME1.getName())).append(Path.SEPARATOR)
               .append(Bytes.toString(loadDescriptor.getEncodedRegionName().toByteArray()))
               .append(Path.SEPARATOR).append(Bytes.toString(FAM_NAME1)).append(Path.SEPARATOR)
-              .append("hfile_" + i).toString();
+              .append(hfilePrefix + i).toString();
       String dst = baseNamespaceDir + Path.SEPARATOR + pathToHfileFromNS;
-
-      FileUtil.copy(fs, p.get(0), fs, new Path(dst), false, conf);
+      Path dstPath = new Path(dst);
+      FileUtil.copy(fs, p.get(0), fs, dstPath, false, conf);
     }
 
     entries.add(builder.build());
@@ -388,6 +388,7 @@ public class TestReplicationSink {
       // 8. Assert data is replicated
       assertEquals(numRows, scanner.next(numRows).length);
     }
+    // Clean up the created hfiles or it will mess up subsequent tests
   }
 
   private WALEntry createEntry(TableName table, int row,  KeyValue.Type type, List<Cell> cells) {