HBASE-28951 add server shortName in temp recovered edits files

Author: ukumawat

hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractRecoveredEditsOutputSink.java

@@ -22,6 +22,8 @@
 
 import java.io.EOFException;
 import java.io.IOException;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -30,8 +32,10 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.log.HBaseMarkers;
+import org.apache.hadoop.hbase.util.Addressing;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -57,11 +61,9 @@ protected RecoveredEditsWriter createRecoveredEditsWriter(TableName tableName, b
     long seqId) throws IOException {
     // If multiple worker are splitting a WAL at a same time, both should use unique file name to
     // avoid conflict
-    long workerStartCode =
-      walSplitter.rsServices != null ? walSplitter.rsServices.getServerName().getStartCode() : 0L;
     Path regionEditsPath = getRegionSplitEditsPath(tableName, region, seqId,
       walSplitter.getFileBeingSplit().getPath().getName(), walSplitter.getTmpDirName(),
-      walSplitter.conf, workerStartCode);
+      walSplitter.conf, getWorkerNameComponent());
 
     if (walSplitter.walFS.exists(regionEditsPath)) {
       LOG.warn("Found old edits file. It could be the "
@@ -78,6 +80,16 @@ protected RecoveredEditsWriter createRecoveredEditsWriter(TableName tableName, b
     return new RecoveredEditsWriter(region, regionEditsPath, w, seqId);
   }
 
+  private String getWorkerNameComponent() {
+    if (walSplitter.rsServices == null) {
+      return "";
+    }
+    return URLEncoder.encode(
+      walSplitter.rsServices.getServerName().toShortString()
+        .replace(Addressing.HOSTNAME_PORT_SEPARATOR, ServerName.SERVERNAME_SEPARATOR),
+      StandardCharsets.UTF_8);
+  }
+
   /**
    * abortRecoveredEditsWriter closes the editsWriter, but does not rename and finalize the
    * recovered edits WAL files. Please see HBASE-28569.
@@ -143,6 +155,8 @@ protected Path closeRecoveredEditsWriterAndFinalizeEdits(RecoveredEditsWriter ed
   private boolean deleteTmpIfDstHasNoLessEntries(RecoveredEditsWriter editsWriter, Path dst)
     throws IOException {
     if (walSplitter.walFS.exists(dst) && !isDstHasFewerEntries(editsWriter, dst)) {
+      LOG.info("Destination {} already have no fewer entries so deleting tmp recovered edits file {}", dst,
+        editsWriter.path);
       if (!walSplitter.walFS.delete(editsWriter.path, false)) {
         LOG.warn("Failed deleting of {}", editsWriter.path);
         throw new IOException("Failed deleting of " + editsWriter.path);

hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitUtil.java

@@ -151,17 +151,18 @@ public static void moveWAL(FileSystem fs, Path p, Path targetDir) throws IOExcep
    * /hbase/some_table/2323432434/recovered.edits/2332. This method also ensures existence of
    * RECOVERED_EDITS_DIR under the region creating it if necessary. And also set storage policy for
    * RECOVERED_EDITS_DIR if WAL_STORAGE_POLICY is configured.
-   * @param tableName          the table name
-   * @param encodedRegionName  the encoded region name
-   * @param seqId              the sequence id which used to generate file name
-   * @param fileNameBeingSplit the file being split currently. Used to generate tmp file name.
-   * @param tmpDirName         of the directory used to sideline old recovered edits file
-   * @param conf               configuration
+   * @param tableName           the table name
+   * @param encodedRegionName   the encoded region name
+   * @param seqId               the sequence id which used to generate file name
+   * @param fileNameBeingSplit  the file being split currently. Used to generate tmp file name.
+   * @param tmpDirName          of the directory used to sideline old recovered edits file
+   * @param conf                configuration
+   * @param workerNameComponent the worker name component for the file name
    * @return Path to file into which to dump split log edits.
    */
   @SuppressWarnings("deprecation")
   static Path getRegionSplitEditsPath(TableName tableName, byte[] encodedRegionName, long seqId,
-    String fileNameBeingSplit, String tmpDirName, Configuration conf, long workerStartCode)
+    String fileNameBeingSplit, String tmpDirName, Configuration conf, String workerNameComponent)
     throws IOException {
     FileSystem walFS = CommonFSUtils.getWALFileSystem(conf);
     Path tableDir = CommonFSUtils.getWALTableDir(conf, tableName);
@@ -195,7 +196,7 @@ static Path getRegionSplitEditsPath(TableName tableName, byte[] encodedRegionNam
     // region's replayRecoveredEdits will not delete it
     String fileName = formatRecoveredEditsFileName(seqId);
     fileName =
-      getTmpRecoveredEditsFileName(fileName + "-" + fileNameBeingSplit + "-" + workerStartCode);
+      getTmpRecoveredEditsFileName(fileName + "-" + fileNameBeingSplit + "-" + workerNameComponent);
     return new Path(dir, fileName);
   }
 

hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALSplit.java

@@ -66,6 +66,8 @@
 import org.apache.hadoop.hbase.coordination.SplitLogWorkerCoordination;
 import org.apache.hadoop.hbase.master.SplitLogManager;
 import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.LastSequenceId;
+import org.apache.hadoop.hbase.regionserver.RegionServerServices;
 import org.apache.hadoop.hbase.regionserver.wal.AbstractProtobufWALReader;
 import org.apache.hadoop.hbase.regionserver.wal.FaultyProtobufWALStreamReader;
 import org.apache.hadoop.hbase.regionserver.wal.InstrumentedLogWriter;
@@ -105,6 +107,7 @@
 import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
 
 import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClusterStatusProtos;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;
 
 /**
@@ -372,6 +375,69 @@ private void loop(final Writer writer) {
     }
   }
 
+  //If another worker is assigned to split a WAl and last worker is still running, both should not impact each other's progress
+  @Test
+  public void testTwoWorkerSplittingSameWAL() throws IOException, InterruptedException {
+    int numWriter = 1, entries = 10;
+    generateWALs(numWriter, entries, -1, 0);
+    FileStatus logfile = fs.listStatus(WALDIR)[0];
+    FileSystem spiedFs = Mockito.spy(fs);
+    RegionServerServices zombieRSServices = Mockito.mock(RegionServerServices.class);
+    RegionServerServices newWorkerRSServices = Mockito.mock(RegionServerServices.class);
+    Mockito.when(zombieRSServices.getServerName())
+      .thenReturn(ServerName.valueOf("zombie-rs.abc.com,1234,1234567890"));
+    Mockito.when(newWorkerRSServices.getServerName())
+      .thenReturn(ServerName.valueOf("worker-rs.abc.com,1234,1234569870"));
+    Thread zombieWorker = new SplitWALWorker(logfile, spiedFs, zombieRSServices);
+    Thread newWorker = new SplitWALWorker(logfile, spiedFs, newWorkerRSServices);
+    zombieWorker.start();
+    newWorker.start();
+    newWorker.join();
+    zombieWorker.join();
+
+    for (String region : REGIONS) {
+      Path[] logfiles = getLogForRegion(TABLE_NAME, region);
+      assertEquals("wrong number of split files for region", numWriter, logfiles.length);
+
+      int count = 0;
+      for (Path lf : logfiles) {
+        count += countWAL(lf);
+      }
+      assertEquals("wrong number of edits for region " + region, entries, count);
+    }
+  }
+
+  private class SplitWALWorker extends Thread implements LastSequenceId {
+    final FileStatus logfile;
+    final FileSystem fs;
+    final RegionServerServices rsServices;
+
+    public SplitWALWorker(FileStatus logfile, FileSystem fs, RegionServerServices rsServices) {
+      super(rsServices.getServerName().toShortString());
+      setDaemon(true);
+      this.fs = fs;
+      this.logfile = logfile;
+      this.rsServices = rsServices;
+    }
+
+    @Override
+    public void run() {
+      try {
+        boolean ret =
+          WALSplitter.splitLogFile(HBASEDIR, logfile, fs, conf, null, this, null, wals, rsServices);
+        assertTrue("Both splitting should pass", ret);
+      } catch (IOException e) {
+        LOG.warn(getName() + " Worker exiting " + e);
+      }
+    }
+
+    @Override
+    public ClusterStatusProtos.RegionStoreSequenceIds getLastSequenceId(byte[] encodedRegionName) {
+      return ClusterStatusProtos.RegionStoreSequenceIds.newBuilder()
+        .setLastFlushedSequenceId(HConstants.NO_SEQNUM).build();
+    }
+  }
+
   /**
    * @see "https://issues.apache.org/jira/browse/HBASE-3020"
    */
@@ -403,7 +469,7 @@ public void testOldRecoveredEditsFileSidelined() throws IOException {
   private Path createRecoveredEditsPathForRegion() throws IOException {
     byte[] encoded = RegionInfoBuilder.FIRST_META_REGIONINFO.getEncodedNameAsBytes();
     Path p = WALSplitUtil.getRegionSplitEditsPath(TableName.META_TABLE_NAME, encoded, 1,
-      FILENAME_BEING_SPLIT, TMPDIRNAME, conf, 0L);
+      FILENAME_BEING_SPLIT, TMPDIRNAME, conf, "");
     return p;
   }